GlcoinCore/lndg
1
0
Fork 0
mirror of https://github.com/cryptosharks131/lndg.git synced 2026-08-13 12:33:04 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Logging Upgrade (#431)

Browse source
This commit is contained in:
cryptosharks131 2025-10-20 12:56:27 -04:00 • committed by GitHub
parent 892211159e
commit 40a8e55d08
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 194 additions and 122 deletions
Download patch file Download diff file Expand all files Collapse all files

11
controller.py
View file

@ -1,12 +1,14 @@
import multiprocessing, sys
import jobs, rebalancer, htlc_stream, p2p, manage
import logging
logger = logging.getLogger('[Controller]')
def run_task(task):
task()
def main():
tasks = [jobs.main, rebalancer.main, htlc_stream.main, p2p.main]
print('Controller is starting...')
logger.info('Starting all LNDg processes...')
processes = []
for task in tasks:
@ -15,15 +17,14 @@ def main():
process.start()
if len(sys.argv) > 1:
sys.argv[0] = "manage.py"
process = multiprocessing.Process(target=manage.main(sys.argv), name="manage.py")
sys.argv[0] = 'manage.py'
process = multiprocessing.Process(target=manage.main(sys.argv), name='manage.py')
processes.append(process)
process.start()
for process in processes:
process.join()
print('Controller is stopping...')
logger.info('Stopping all LNDg processes...')
if __name__ == '__main__':
main()

29
gui/views.py
View file

@ -29,6 +29,8 @@ from requests import get
from secrets import token_bytes
from trade import create_trade_details
import af
import logging
logger = logging.getLogger('django.lndg')
def graph_links():
if LocalSettings.objects.filter(key='GUI-GraphLinks').exists():
@ -409,12 +411,10 @@ def closures(request):
return redirect('home')
def find_next_block_maturity(force_closing_channel):
#print (f"{datetime.now().strftime('%c')} : {force_closing_channel=}")
if force_closing_channel.blocks_til_maturity > 0:
return force_closing_channel.blocks_til_maturity
for pending_htlc in force_closing_channel.pending_htlcs:
if pending_htlc.blocks_til_maturity > 0:
#print (f"{datetime.now().strftime('%c')} : {pending_htlc=}")
return pending_htlc.blocks_til_maturity
return -1
@ -1113,7 +1113,7 @@ def unprofitable_channels(request):
stub = lnrpc.LightningStub(lnd_connect())
current_block_height = stub.GetInfo(ln.GetInfoRequest()).block_height
except Exception as e:
print(f"Error getting current block height: {e}")
logger.error(f'Error getting current block height: {str(e)}')
current_block_height = 0
VERY_NEW_DAYS = 7
@ -1350,7 +1350,7 @@ def unprofitable_channels(request):
except Exception as e:
# Handle potential errors during age calculation gracefully
print(f"Error calculating age for chan_id {channel.chan_id}: {e}")
logger.error(f'Error calculating age for chan_id {channel.chan_id}: {str(e)}')
channel_age_days = -1 # Indicate unknown age
else:
channel_age_days = -1 # Indicate unknown age if block height or chan_id missing
@ -1429,26 +1429,25 @@ def actions(request):
result['auto_rebalance'] = channel.auto_rebalance
result['ar_target'] = channel.ar_in_target
if result['o7D'] > (result['i7D']*1.10) and result['outbound_percent'] > 75:
#print('Case 1: Pass')
continue
logger.debug('Auto-Enable Case 1: Pass')
elif result['o7D'] > (result['i7D']*1.10) and result['inbound_percent'] > 75 and channel.auto_rebalance == False:
if channel.local_fee_rate <= channel.remote_fee_rate:
#print('Case 6: Peer Fee Too High')
logger.debug('Case 6: Peer Fee Too High')
result['output'] = 'Peer Fee Too High'
result['reason'] = 'o7D > i7D AND Inbound Liq > 75% AND Local Fee < Remote Fee'
continue
#print('Case 2: Enable AR')
logger.debug('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%')
result['output'] = 'Enable AR'
result['reason'] = 'o7D > i7D AND Inbound Liq > 75%'
elif result['o7D'] < (result['i7D']*1.10) and result['outbound_percent'] > 75 and channel.auto_rebalance == True:
#print('Case 3: Disable AR')
logger.debug('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%')
result['output'] = 'Disable AR'
result['reason'] = 'o7D < i7D AND Outbound Liq > 75%'
elif result['o7D'] < (result['i7D']*1.10) and result['inbound_percent'] > 75:
#print('Case 4: Pass')
logger.debug('Case 4: Pass')
continue
else:
#print('Case 5: Pass')
logger.debug('Case 5: Pass')
continue
if len(result) > 0:
action_list.append(result)
@ -1725,7 +1724,7 @@ def batch_open(request):
channel_open.local_funding_amount = open['amt']
channels.append(channel_open)
response = stub.BatchOpenChannel(ln.BatchOpenChannelRequest(channels=channels, sat_per_vbyte=form.cleaned_data['fee_rate']))
print(response)
logger.debug(f'Batch open response: {response}')
messages.success(request, 'Batch opened channels!')
except Exception as e:
error = str(e)
@ -2777,7 +2776,7 @@ def get_channeldb_file_size():
# Check for required settings
if not host_value or not user_value:
print("Error: Remote file size enabled, but host or user is not set.")
logger.error('Error: Remote file size enabled, but host or user is not set')
return round(path.getsize(path.expanduser(settings.LND_DATABASE_PATH))*0.000000001, 3)
# --- Paramiko logic ---
@ -2805,7 +2804,7 @@ def get_channeldb_file_size():
return round(file_size_bytes * 0.000000001, 3)
except Exception as e:
print(f"Error retrieving file size with paramiko: {e}")
logger.error(f'Error retrieving file size with paramiko: {str(e)}')
return round(path.getsize(path.expanduser(settings.LND_DATABASE_PATH))*0.000000001, 3) # Fallback
# --- End Paramiko logic ---
@ -2815,7 +2814,7 @@ def get_channeldb_file_size():
except Exception as e:
# Handle exceptions
print(f"Error retrieving channel.db file size: {e}")
logger.error(f'Error retrieving channel.db file size: {str(e)}')
return 0
@api_view(['GET'])

7
htlc_stream.py
View file

@ -1,5 +1,4 @@
import django
from datetime import datetime
from gui.lnd_deps import router_pb2 as lnr
from gui.lnd_deps import router_pb2_grpc as lnrouter
from gui.lnd_deps.lnd_connect import lnd_connect
@ -8,11 +7,13 @@ from time import sleep
environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings'
django.setup()
from gui.models import Channels, FailedHTLCs
import logging
logger = logging.getLogger('[HTLC]')
def main():
while True:
try:
print(f"{datetime.now().strftime('%c')} : [HTLC] : Starting failed HTLC stream...")
logger.info('Starting failed HTLC stream...')
connection = lnd_connect()
routerstub = lnrouter.RouterStub(connection)
all_forwards = {}
@ -59,7 +60,7 @@ def main():
FailedHTLCs(amount=amount, chan_id_in=in_chan_id, chan_id_out=out_chan_id, chan_in_alias=in_chan_alias, chan_out_alias=out_chan_alias, chan_out_liq=out_chan_liq, chan_out_pending=out_chan_pending, wire_failure=wire_failure, failure_detail=failure_detail, missed_fee=missed_fee).save()
del all_forwards[key]
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [HTLC] : Error while running failed HTLC stream: {str(e)}")
logger.error(f'Error while running failed HTLC stream: {str(e)}')
sleep(20)
if __name__ == '__main__':

70
initialize.py
View file

@ -171,6 +171,76 @@ STATIC_URL = 'static/'
STATIC_ROOT = os.path.join(BASE_DIR, 'gui/static/')
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
SESSION_COOKIE_AGE = %s
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{asctime} {name} {levelname} - {message}',
'style': '{',
},
'simple': {
'format': '{asctime} {name} - {message}',
'style': '{',
},
},
'handlers': {
'app-file': {
'level': 'INFO',
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'data/lndg-controller.log',
'maxBytes': 25*(1024*1024),
'backupCount': 5,
'formatter': 'verbose',
},
'web-file': {
'level': 'INFO',
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'data/lndg-web.log',
'maxBytes': 25*(1024*1024),
'backupCount': 5,
'formatter': 'verbose',
},
'console': {
'level': 'INFO',
'class': 'logging.StreamHandler',
'formatter': 'simple',
},
},
'loggers': {
'[Controller]': {
'handlers': ['app-file', 'console'],
'level': 'INFO',
'propagate': False,
},
'[Data]': {
'handlers': ['app-file', 'console'],
'level': 'INFO',
'propagate': False,
},
'[Rebalancer]': {
'handlers': ['app-file', 'console'],
'level': 'INFO',
'propagate': False,
},
'[HTLC]': {
'handlers': ['app-file', 'console'],
'level': 'INFO',
'propagate': False,
},
'[P2P]': {
'handlers': ['app-file', 'console'],
'level': 'INFO',
'propagate': False,
},
'django': {
'handlers': ['web-file', 'console'],
'level': 'INFO',
'propagate': False,
},
},
}
''' % (secret, debug, node_ip, csrf, lnd_tls_path, lnd_macaroon_path, lnd_database_path, lnd_network, lnd_rpc_server, lnd_max_message, not nologinrequired, wnl, api_login, cookie_age)
if not force_new and Path("lndg/settings.py").exists():
print('A settings file already exist, skipping creation...')

71
jobs.py
View file

@ -15,7 +15,8 @@ environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings'
django.setup()
from gui.models import Payments, PaymentHops, Invoices, Forwards, Channels, Peers, Onchain, Closures, Resolutions, PendingHTLCs, LocalSettings, FailedHTLCs, Autofees, InboundFeeLog, PendingChannels, HistFailedHTLC, PeerEvents
import af
import logging
logger = logging.getLogger('[Data]')
_SELF_PUBKEY = None
def _get_self_pubkey(stub):
@ -24,7 +25,7 @@ def _get_self_pubkey(stub):
try:
_SELF_PUBKEY = stub.GetInfo(ln.GetInfoRequest()).identity_pubkey
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : GetInfo failed: {e}")
logger.critical(f'GetInfo failed to get self pubkey: {e}')
raise
return _SELF_PUBKEY
@ -33,7 +34,7 @@ def update_payments(stub):
inflight_payments = Payments.objects.filter(status=1).order_by('index')
for payment in inflight_payments:
payment_data = stub.ListPayments(ln.ListPaymentsRequest(include_incomplete=True, index_offset=payment.index-1, max_payments=1)).payments
#Ignore inflight payments before 30 days
# Ignore inflight payments before 30 days
if len(payment_data) > 0 and payment.payment_hash == payment_data[0].payment_hash and payment.creation_date > (datetime.now() - timedelta(days=30)):
update_payment(stub, payment_data[0], self_pubkey)
else:
@ -79,7 +80,7 @@ def update_payments(stub):
try:
Payments.objects.bulk_create(new_payments, ignore_conflicts=True, batch_size=1000)
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error bulk inserting payments: {str(e)}")
logger.error(f'Error bulk inserting payments: {str(e)}')
for p in payments:
if (p.payment_hash in new_hashes) or (p.status in (0, 1)):
@ -127,7 +128,7 @@ def update_payments(stub):
try:
Payments.objects.bulk_create(new_payments, ignore_conflicts=True, batch_size=1000)
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error bulk inserting payments: {str(e)}")
logger.error(f'Error bulk inserting payments: {str(e)}')
for p in payments:
if (p.payment_hash in new_hashes) or (p.status in (0, 1)):
@ -214,7 +215,7 @@ def update_invoice(stub, invoice, db_invoice):
try:
valid = signerstub.VerifyMessage(lns.VerifyMessageReq(msg=(records[34349339]+bytes.fromhex(self_pubkey)+records[34349343]+records[34349334]), signature=records[34349337], pubkey=records[34349339])).valid
except:
print(f"{datetime.now().strftime('%c')} : [Data] : Unable to validate signature on invoice: {invoice.r_hash.hex()}")
logger.error(f'Unable to validate signature on invoice: {invoice.r_hash.hex()}')
valid = False
sender = records[34349339].hex() if valid == True else None
try:
@ -286,11 +287,11 @@ def update_forwards(stub):
def disconnectpeer(stub, peer):
try:
stub.DisconnectPeer(ln.DisconnectPeerRequest(pub_key=peer.pubkey))
print(f"{datetime.now().strftime('%c')} : [Data] : Disconnected peer {peer.alias} {peer.pubkey}")
logger.info(f'Disconnected peer {peer.alias} {peer.pubkey}')
peer.connected = False
peer.save()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error disconnecting peer {peer.alias} {peer.pubkey}: {str(e)}")
logger.error(f'Error disconnecting peer {peer.alias} {peer.pubkey}: {str(e)}')
def update_channels(stub):
counter = 0
@ -302,11 +303,11 @@ def update_channels(stub):
version = get_info.version
for channel in channels:
if Channels.objects.filter(chan_id=channel.chan_id).exists():
#Update the channel record with the most current data
# Update the channel record with the most current data
db_channel = Channels.objects.filter(chan_id=channel.chan_id)[0]
pending_channel = None
else:
#Create a record for this new channel
# Create a record for this new channel
try:
alias = stub.GetNodeInfo(ln.NodeInfoRequest(pub_key=channel.remote_pubkey, include_channels=False)).node.alias
except:
@ -359,12 +360,12 @@ def update_channels(stub):
if htlc.expiration_height - block_height <= 13: # If htlc is expiring within 13 blocks, disconnect peer to help resolve the stuck htlc
peer = Peers.objects.filter(pubkey=channel.remote_pubkey)[0] if Peers.objects.filter(pubkey=channel.remote_pubkey).exists() else None
if peer and (not peer.last_reconnected or (int((datetime.now() - peer.last_reconnected).total_seconds() / 60) > 10)):
print(f"{datetime.now().strftime('%c')} : [Data] : HTLC expiring at {htlc.expiration_height} and within 13 blocks of {block_height}, disconnecting peer {channel.remote_pubkey} to resolve HTLC: {htlc.hash_lock.hex()} ")
logger.info(f'HTLC expiring at {htlc.expiration_height} and within 13 blocks of {block_height}, disconnecting peer {channel.remote_pubkey} to resolve HTLC: {htlc.hash_lock.hex()}')
disconnectpeer(stub, peer)
peer.last_reconnected = datetime.now()
peer.save()
else:
print(f"{datetime.now().strftime('%c')} : [Data] : Could not find peer {channel.remote_pubkey} with expiring HTLC: {htlc.hash_lock.hex()}")
logger.error(f'Could not find peer {channel.remote_pubkey} with expiring HTLC: {htlc.hash_lock.hex()}')
db_channel.pending_outbound = pending_out
db_channel.pending_inbound = pending_in
db_channel.htlc_count = htlc_counter
@ -474,9 +475,9 @@ def update_channels(stub):
db_channel.remote_inbound_base_fee = 0
db_channel.remote_inbound_fee_rate = 0
except Exception as e: # LND has not found the channel on the graph
print(f"{datetime.now().strftime('%c')} : [Data] : Error getting graph data for channel {db_channel.chan_id}: {str(e)}")
logger.error(f'Error getting graph data for channel {db_channel.chan_id}: {str(e)}')
if pending_channel: # skip adding new channel to the list, LND may not have added to the graph yet
print(f"{datetime.now().strftime('%c')} : [Data] : Waiting for pending channel {db_channel.chan_id} to be added to the graph...")
logger.error(f'Waiting for pending channel {db_channel.chan_id} to be added to the graph...')
continue
else:
old_fee_rate = None
@ -525,16 +526,16 @@ def update_channels(stub):
db_channel.auto_fees = pending_channel.auto_fees
pending_channel.delete()
if old_fee_rate is not None and old_fee_rate != local_policy.fee_rate_milli_msat:
print(f"{datetime.now().strftime('%c')} : [Data] : Ext fee change detected on {db_channel.chan_id} for peer {db_channel.alias}: fee updated from {old_fee_rate} to {db_channel.local_fee_rate}")
#External Fee change detected, update auto fee log
logger.info(f'Ext fee change detected on {db_channel.chan_id} for peer {db_channel.alias}: fee updated from {old_fee_rate} to {db_channel.local_fee_rate}')
# External Fee change detected, update auto fee log
db_channel.fees_updated = datetime.now()
Autofees(chan_id=db_channel.chan_id, peer_alias=db_channel.alias, setting=(f"Ext"), old_value=old_fee_rate, new_value=db_channel.local_fee_rate).save()
Autofees(chan_id=db_channel.chan_id, peer_alias=db_channel.alias, setting=('Ext'), old_value=old_fee_rate, new_value=db_channel.local_fee_rate).save()
db_channel.save()
counter += 1
chan_list.append(channel.chan_id)
records = Channels.objects.filter(is_open=True).count()
if records > counter:
#A channel must have been closed, mark it as closed
# A channel must have been closed, mark it as closed
channels = Channels.objects.filter(is_open=True).exclude(chan_id__in=chan_list)
for channel in channels:
channel.last_update = datetime.now()
@ -599,7 +600,7 @@ def get_tx_fees(txid):
request_data = get(base_url + txid).json()
fee = request_data['fee']
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error getting closure fees for {txid}: {str(e)}")
logger.error(f'Error getting closure fees for {txid}: {str(e)}')
fee = 0
return fee
@ -618,7 +619,7 @@ def update_closures(stub):
try:
db_closure.save()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error inserting closure: {str(e)}")
logger.error(f'Error inserting closure: {str(e)}')
Closures.objects.filter(funding_txid=txid,funding_index=index).delete()
return
if resolution_count > 0:
@ -638,19 +639,19 @@ def reconnect_peers(stub):
if peers.filter(pubkey=inactive_peer).exists():
peer = peers.filter(pubkey=inactive_peer)[0]
if peer.last_reconnected == None or (int((datetime.now() - peer.last_reconnected).total_seconds() / 60) > 2):
print(f"{datetime.now().strftime('%c')} : [Data] : Reconnecting peer {peer.alias} {peer.pubkey}, last reconnected at {peer.last_reconnected}")
logger.info(f'Reconnecting peer {peer.alias} {peer.pubkey}, last reconnected at {peer.last_reconnected}')
if peer.connected == True:
print(f"{datetime.now().strftime('%c')} : [Data] : Inactive channel is still connected to peer, disconnecting peer {peer.alias} {inactive_peer}")
logger.info(f'Inactive channel is still connected to peer, disconnecting peer {peer.alias} {inactive_peer}')
disconnectpeer(stub, peer)
try:
node = stub.GetNodeInfo(ln.NodeInfoRequest(pub_key=inactive_peer, include_channels=False)).node
host = node.addresses[0].addr
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Unable to find node info on graph, using last known value for {peer.alias} {peer.pubkey} at {peer.address}: {str(e)}")
logger.error(f'Unable to find node info on graph, using last known value for {peer.alias} {peer.pubkey} at {peer.address}: {str(e)}')
host = peer.address
print(f"{datetime.now().strftime('%c')} : [Data] : Attempting connection to {peer.alias} {inactive_peer} at {host}")
logger.info(f'Attempting connection to {peer.alias} {inactive_peer} at {host}')
try:
#try both the graph value and last know value
# try both the graph value and last know value
stub.ConnectPeer(request = ln.ConnectPeerRequest(addr=ln.LightningAddress(pubkey=inactive_peer, host=host), perm=True, timeout=5))
if host != peer.address and peer.address[:9] != '127.0.0.1':
stub.ConnectPeer(request = ln.ConnectPeerRequest(addr=ln.LightningAddress(pubkey=inactive_peer, host=peer.address), perm=True, timeout=5))
@ -659,7 +660,7 @@ def reconnect_peers(stub):
details_index = error.find('details =') + 11
debug_error_index = error.find('debug_error_string =') - 3
error_msg = error[details_index:debug_error_index]
print(f"{datetime.now().strftime('%c')} : [Data] : Error reconnecting {peer.alias} {inactive_peer}: {error_msg}")
logger.error(f'Error reconnecting {peer.alias} {inactive_peer}: {error_msg}')
peer.last_reconnected = datetime.now()
peer.save()
@ -687,7 +688,7 @@ def clean_payments(stub):
details_index = error.find('details =') + 11
debug_error_index = error.find('debug_error_string =') - 3
error_msg = error[details_index:debug_error_index]
print(f"{datetime.now().strftime('%c')} : [Data] : Error cleaning payment {payment.payment_hash} at index {payment.index} with payment status {payment.status}: {error_msg}")
logger.error(f'Error cleaning payment {payment.payment_hash} at index {payment.index} with payment status {payment.status}: {error_msg}')
finally:
payment.cleaned = True
payment.save()
@ -727,19 +728,19 @@ def auto_fees(stub):
inbound_base_fee = -channel.local_base_fee
stub.UpdateChannelPolicy(ln.PolicyUpdateRequest(chan_point=channel_point, base_fee_msat=channel.local_base_fee, fee_rate=(target_channel['new_rate']/1000000), time_lock_delta=channel.local_cltv, inbound_fee=ln.InboundFee(base_fee_msat=inbound_base_fee, fee_rate_ppm=inbound_fee_rate)))
if target_channel['inbound_adjustment'] != 0:
print(f"{datetime.now().strftime('%c')} : [Data] : Updating inbound fees for channel {str(target_channel['chan_id'])} to a value of: {str(target_channel['new_inbound_rate'])}")
logger.info(f'Updating inbound fees for channel {str(target_channel["chan_id"])} to a value of: {str(target_channel["new_inbound_rate"])}')
channel.local_inbound_fee_rate = target_channel['new_inbound_rate']
InboundFeeLog(chan_id=channel.chan_id, peer_alias=channel.alias, setting=(f"AF [ {target_channel['net_routed_7day']}:{target_channel['in_percent']}:{target_channel['out_percent']} ]"), old_value=target_channel['local_inbound_fee_rate'], new_value=target_channel['new_inbound_rate']).save()
else:
stub.UpdateChannelPolicy(ln.PolicyUpdateRequest(chan_point=channel_point, base_fee_msat=channel.local_base_fee, fee_rate=(target_channel['new_rate']/1000000), time_lock_delta=channel.local_cltv))
if target_channel['adjustment'] != 0:
print(f"{datetime.now().strftime('%c')} : [Data] : Updating outbound fees for channel {str(target_channel['chan_id'])} to a value of: {str(target_channel['new_rate'])}")
logger.info(f'Updating outbound fees for channel {str(target_channel["chan_id"])} to a value of: {str(target_channel["new_rate"])}')
channel.local_fee_rate = target_channel['new_rate']
Autofees(chan_id=channel.chan_id, peer_alias=channel.alias, setting=(f"AF [ {target_channel['net_routed_7day']}:{target_channel['in_percent']}:{target_channel['out_percent']} ]"), old_value=target_channel['local_fee_rate'], new_value=target_channel['new_rate']).save()
channel.fees_updated = datetime.now()
channel.save()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error processing auto_fees: {str(e)}")
logger.error(f'Error processing auto_fees: {str(e)}')
def agg_htlcs(target_htlcs, category):
@ -770,7 +771,7 @@ def agg_htlcs(target_htlcs, category):
htlc_itm.save()
FailedHTLCs.objects.filter(id__in=target_ids, chan_id_in=htlc['chan_id_in'], chan_id_out=htlc['chan_id_out']).annotate(day=TruncDay('timestamp')).filter(day=htlc['day']).delete()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error processing agg_htlcs: {str(e)}")
logger.error(f'Error processing agg_htlcs: {str(e)}')
def agg_failed_htlcs():
time_filter = datetime.now() - timedelta(days=30)
@ -780,10 +781,10 @@ def agg_failed_htlcs():
def main():
while True:
print(f"{datetime.now().strftime('%c')} : [Data] : Starting data execution...")
logger.info('Starting data execution...')
try:
stub = lnrpc.LightningStub(lnd_connect())
#Update data
# Update data
update_peers(stub)
update_channels(stub)
update_invoices(stub)
@ -796,8 +797,8 @@ def main():
auto_fees(stub)
agg_failed_htlcs()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Data] : Error processing background data: {str(e)}")
print(f"{datetime.now().strftime('%c')} : [Data] : Data execution completed...sleeping for 20 seconds")
logger.error(f'Error processing background data: {str(e)}')
logger.info('Data execution completed...sleeping for 20 seconds')
sleep(20)
if __name__ == '__main__':

11
p2p.py
View file

@ -1,5 +1,4 @@
import django, multiprocessing
from datetime import datetime
from gui.lnd_deps import lightning_pb2_grpc as lnrpc
from gui.lnd_deps.lnd_connect import lnd_connect
from os import environ
@ -8,6 +7,8 @@ environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings'
django.setup()
from gui.models import LocalSettings
from trade import serve_trades
import logging
logger = logging.getLogger('[P2P]')
def trade():
stub = lnrpc.LightningStub(lnd_connect())
@ -28,21 +29,21 @@ def main():
db_value = check_setting()
if current_value != db_value:
if db_value == 1:
print(f"{datetime.now().strftime('%c')} : [P2P] : Starting the p2p service...")
logger.info('Starting the p2p service...')
django.db.connections.close_all()
p2p_thread = multiprocessing.Process(target=trade)
p2p_thread.start()
else:
if 'p2p_thread' in locals() and p2p_thread.is_alive():
print(f"{datetime.now().strftime('%c')} : [P2P] : Stopping the p2p service...")
logger.info('Stopping the p2p service...')
p2p_thread.terminate()
current_value = db_value
sleep(2) # polling interval
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [P2P] : Error running p2p service: {str(e)}")
logger.error(f'Error running p2p service: {str(e)}')
finally:
if 'p2p_thread' in locals() and p2p_thread.is_alive():
print(f"{datetime.now().strftime('%c')} : [P2P] : Removing any remaining processes...")
logger.info('Removing any remaining processes...')
p2p_thread.terminate()
sleep(20)

97
rebalancer.py
View file

@ -10,31 +10,32 @@ from gui.lnd_deps import router_pb2_grpc as lnrouter
from gui.lnd_deps.lnd_connect import lnd_connect, async_lnd_connect
from os import environ
from typing import List
environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings'
django.setup()
from gui.models import Rebalancer, Channels, LocalSettings, Forwards, Autopilot
import logging
logger = logging.getLogger('[Rebalancer]')
@sync_to_async
def get_out_cans(rebalance, auto_rebalance_channels):
try:
return list(auto_rebalance_channels.filter(auto_rebalance=False, percent_outbound__gte=F('ar_out_target')).exclude(remote_pubkey=rebalance.last_hop_pubkey).values_list('chan_id', flat=True))
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting outbound cands: {str(e)}")
logger.error(f'Error getting outbound cands: {str(e)}')
@sync_to_async
def save_record(record):
try:
record.save()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error saving database record: {str(e)}")
logger.error(f'Error saving database record: {str(e)}')
@sync_to_async
def inbound_cans_len(inbound_cans):
try:
return len(inbound_cans)
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting inbound cands: {str(e)}")
logger.error(f'Error getting inbound cands: {str(e)}')
@sync_to_async
def check_and_set_allow_multishards():
@ -59,7 +60,7 @@ async def run_rebalancer(rebalance, worker):
auto_rebalance_channels = Channels.objects.filter(is_active=True, is_open=True, private=False).annotate(percent_outbound=((Sum('local_balance')+Sum('pending_outbound')-rebalance.value)*100)/Sum('capacity')).annotate(inbound_can=(((Sum('remote_balance')+Sum('pending_inbound'))*100)/Sum('capacity'))/Sum('ar_in_target'))
outbound_cans = await get_out_cans(rebalance, auto_rebalance_channels)
if len(outbound_cans) == 0 and rebalance.manual == False:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : No outbound_cans")
logger.info('No outbound_cans')
rebalance.status = 406
rebalance.start = datetime.now()
rebalance.stop = datetime.now()
@ -75,7 +76,7 @@ async def run_rebalancer(rebalance, worker):
chan_ids = json.loads(rebalance.outgoing_chan_ids)
timeout = rebalance.duration * 60
invoice_response = stub.AddInvoice(ln.Invoice(value=rebalance.value, expiry=timeout))
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} starting rebalance for {rebalance.target_alias} {rebalance.last_hop_pubkey} for {rebalance.value} sats and duration {rebalance.duration}, using {len(chan_ids)} outbound channels")
logger.debug(f'{worker} starting rebalance for {rebalance.target_alias} {rebalance.last_hop_pubkey} for {rebalance.value} sats and duration {rebalance.duration}, using {len(chan_ids)} outbound channels')
async for payment_response in routerstub.SendPaymentV2(lnr.SendPaymentRequest(payment_request=str(invoice_response.payment_request), fee_limit_msat=int(rebalance.fee_limit*1000), outgoing_chan_ids=chan_ids, last_hop_pubkey=bytes.fromhex(rebalance.last_hop_pubkey), timeout_seconds=(timeout-5), allow_self_payment=True, max_parts=max_parts), timeout=(timeout+60)):
if payment_response.status == 1 and rebalance.status == 0:
#IN-FLIGHT
@ -111,11 +112,11 @@ async def run_rebalancer(rebalance, worker):
rebalance.status = 408
else:
rebalance.status = 400
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error while sending payment: {str(e)}")
logger.error(f'Error while sending payment: {str(e)}')
finally:
rebalance.stop = datetime.now()
await save_record(rebalance)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} completed payment attempts for: {rebalance.payment_hash}")
logger.debug(f'{worker} completed payment attempts for: {rebalance.payment_hash}')
original_alias = rebalance.target_alias
inc=1.21
dec=2
@ -128,7 +129,7 @@ async def run_rebalancer(rebalance, worker):
if await inbound_cans_len(inbound_cans) > 0 and len(outbound_cans) > 0:
next_rebalance = Rebalancer(value=int(rebalance.value*inc), fee_limit=round(rebalance.fee_limit*inc, 3), outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=rebalance.last_hop_pubkey, target_alias=original_alias, duration=1)
await save_record(next_rebalance)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : RapidFire increase for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}")
logger.info(f'RapidFire increase for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}')
else:
next_rebalance = None
# For failed rebalances, try in rapid fire with reduced balances until give up.
@ -146,14 +147,14 @@ async def run_rebalancer(rebalance, worker):
if await inbound_cans_len(inbound_cans) > 0 and len(outbound_cans) > 0:
next_rebalance = Rebalancer(value=int(next_value), fee_limit=round(rebalance.fee_limit/(rebalance.value/next_value), 3), outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=rebalance.last_hop_pubkey, target_alias=original_alias, duration=1)
await save_record(next_rebalance)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : RapidFire decrease for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}")
logger.info(f'RapidFire decrease for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}')
else:
next_rebalance = None
else:
next_rebalance = None
return next_rebalance
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error running rebalance attempt: {str(e)}")
logger.error(f'Error running rebalance attempt: {str(e)}')
@sync_to_async
def estimate_liquidity( payment ):
@ -166,9 +167,9 @@ def estimate_liquidity( payment ):
if attempt.failure.failure_source_index == total_hops:
#Failure from last hop indicating liquidity available
estimated_liquidity = attempt.route.total_amt if attempt.route.total_amt > estimated_liquidity else estimated_liquidity
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Estimated Liquidity {estimated_liquidity} for payment {payment.payment_hash} with status {payment.status} and reason {payment.failure_reason}")
logger.info(f'Estimated Liquidity {estimated_liquidity} for payment {payment.payment_hash} with status {payment.status} and reason {payment.failure_reason}')
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error estimating liquidity: {str(e)}")
logger.error(f'Error estimating liquidity: {str(e)}')
estimated_liquidity = 0
return estimated_liquidity
@ -189,7 +190,7 @@ def update_channels(stub, incoming_channel, outgoing_channel):
db_channel.remote_balance = channel.remote_balance
db_channel.save()
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error updating channel balances: {str(e)}")
logger.error(f'Error updating channel balances: {str(e)}')
@sync_to_async
def auto_schedule() -> List[Rebalancer]:
@ -255,15 +256,15 @@ def auto_schedule() -> List[Rebalancer]:
last_rebalance = Rebalancer.objects.filter(last_hop_pubkey=target.remote_pubkey).exclude(status=0).order_by('-id')[0]
if not (last_rebalance.status == 2 or (last_rebalance.status > 2 and (int((datetime.now() - last_rebalance.stop).total_seconds() / 60) > wait_period)) or (last_rebalance.status == 1 and ((int((datetime.now() - last_rebalance.start).total_seconds() / 60) - last_rebalance.duration) > wait_period))):
continue
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Creating Auto Rebalance Request for: {target.chan_id}")
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Value: {target_value} / {target.ar_amt_target} | Fee: {target_fee} | Duration: {target_time}")
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Request routing outbound via: {outbound_cans}")
logger.info(f'Creating Auto Rebalance Request for: {target.chan_id}')
logger.info(f'Value: {target_value} / {target.ar_amt_target} | Fee: {target_fee} | Duration: {target_time}')
logger.info(f'Request routing outbound via: {outbound_cans}')
new_rebalance = Rebalancer(value=target_value, fee_limit=target_fee, outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=target.remote_pubkey, target_alias=target.alias, duration=target_time)
new_rebalance.save()
to_schedule.append(new_rebalance)
return to_schedule
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error scheduling rebalances: {str(e)}")
logger.error(f'Error scheduling rebalances: {str(e)}')
return to_schedule
@sync_to_async
@ -294,32 +295,28 @@ def auto_enable():
oapD = 0 if routed_out_apday == 0 else int(forwards.filter(chan_id_out__in=chan_list).aggregate(Sum('amt_out_msat'))['amt_out_msat__sum']/10000000)/100
for peer_channel in lookup_channels.filter(chan_id__in=chan_list):
if peer_channel.ar_out_target == 100 and peer_channel.auto_rebalance == True:
#Special Case for LOOP, Wos, etc. Always Auto Rebalance if enabled to keep outbound full.
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Skipping AR enabled and 100% oTarget channel: {peer_channel.alias} {peer_channel.chan_id}")
pass
logger.debug('Special case for sinks like LOOP, Wos, etc. if AR enabled and oTarget at 100%: Pass')
logger.info(f'Skipping AR enabled and 100% oTarget channel: {peer_channel.alias} {peer_channel.chan_id}')
elif oapD > (iapD*1.10) and outbound_percent > 75:
#print('Case 1: Pass')
pass
logger.debug('Auto-Enable Case 1: Pass')
elif oapD > (iapD*1.10) and inbound_percent > 75 and peer_channel.auto_rebalance == False:
#print('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%')
logger.debug('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%')
peer_channel.auto_rebalance = True
peer_channel.save()
Autopilot(chan_id=peer_channel.chan_id, peer_alias=peer_channel.alias, setting='Enabled', old_value=0, new_value=1).save()
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Auto Pilot Enabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}")
logger.info(f'Auto Pilot Enabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}')
elif oapD < (iapD*1.10) and outbound_percent > 75 and peer_channel.auto_rebalance == True:
#print('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%')
logger.debug('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%')
peer_channel.auto_rebalance = False
peer_channel.save()
Autopilot(chan_id=peer_channel.chan_id, peer_alias=peer_channel.alias, setting='Enabled', old_value=1, new_value=0).save()
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Auto Pilot Disabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}" )
logger.info(f'Auto Pilot Disabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}')
elif oapD < (iapD*1.10) and inbound_percent > 75:
#print('Case 4: Pass')
pass
logger.debug('Case 4: Pass')
else:
#print('Case 5: Pass')
pass
logger.debug('Case 5: Pass')
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error during auto channel enabling: {str(e)}")
logger.error(f'Error during auto channel enabling: {str(e)}')
@sync_to_async
def get_pending_rebals():
@ -327,49 +324,49 @@ def get_pending_rebals():
rebalances = Rebalancer.objects.filter(status=0).order_by('id')
return rebalances, len(rebalances)
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting pending rebalances: {str(e)}")
logger.error(f'Error getting pending rebalances: {str(e)}')
async def async_queue_manager(rebalancer_queue):
global scheduled_rebalances, active_rebalances, shutdown_rebalancer
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager is starting...")
logger.debug('Queue manager is starting...')
try:
while True:
if shutdown_rebalancer == True:
return
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue currently has {rebalancer_queue.qsize()} items...")
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : There are currently {len(active_rebalances)} tasks in progress...")
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager is checking for more work...")
logger.info(f'Queue currently has {rebalancer_queue.qsize()} items...') if rebalancer_queue.qsize() > 0 else logger.debug('Queue currently has no items...')
logger.info(f'There are currently {len(active_rebalances)} tasks in progress...') if len(active_rebalances) > 0 else logger.debug('There are currently no tasks in progress...')
logger.info('Queue manager is checking for more work...')
pending_rebalances, rebal_count = await get_pending_rebals()
if rebal_count > 0:
for rebalance in pending_rebalances:
if rebalance.id not in (scheduled_rebalances + active_rebalances):
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Found a pending job to schedule with id: {rebalance.id}")
logger.info(f'Found a pending job to schedule with id: {rebalance.id}')
scheduled_rebalances.append(rebalance.id)
await rebalancer_queue.put(rebalance)
await auto_enable()
scheduled = await auto_schedule()
if len(scheduled) > 0:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Scheduling {len(scheduled)} more jobs...")
logger.info(f'Scheduling {len(scheduled)} more jobs...')
for rebalance in scheduled:
scheduled_rebalances.append(rebalance.id)
await rebalancer_queue.put(rebalance)
elif rebalancer_queue.qsize() == 0 and len(active_rebalances) == 0:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue is still empty, stopping the rebalancer...")
logger.info('No active work found, stopping the rebalancer...')
shutdown_rebalancer = True
return
await asyncio.sleep(30)
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager exception: {str(e)}")
logger.error(f'Queue manager exception: {str(e)}')
shutdown_rebalancer = True
finally:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager has shut down...")
logger.debug('Queue manager has shut down...')
async def async_run_rebalancer(worker, rebalancer_queue):
global scheduled_rebalances, active_rebalances, shutdown_rebalancer
while True:
if not rebalancer_queue.empty() and not shutdown_rebalancer:
rebalance = await rebalancer_queue.get()
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} is starting a new request...")
logger.debug(f'{worker} is starting a new request...')
active_rebalance_id = None
if rebalance != None:
active_rebalance_id = rebalance.id
@ -379,7 +376,7 @@ async def async_run_rebalancer(worker, rebalancer_queue):
rebalance = await run_rebalancer(rebalance, worker)
if active_rebalance_id != None:
active_rebalances.remove(active_rebalance_id)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} completed its request...")
logger.debug(f'{worker} completed its request...')
else:
if shutdown_rebalancer == True:
return
@ -390,7 +387,7 @@ async def start_queue(worker_count=1):
manager = asyncio.create_task(async_queue_manager(rebalancer_queue))
workers = [asyncio.create_task(async_run_rebalancer("Worker " + str(worker_num+1), rebalancer_queue)) for worker_num in range(worker_count)]
await asyncio.gather(manager, *workers)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Manager and workers have stopped...")
logger.debug('Manager and workers have stopped...')
@sync_to_async
def get_worker_count():
@ -406,7 +403,7 @@ async def update_worker_count():
if updated_worker_count != worker_count:
worker_count = updated_worker_count
shutdown_rebalancer = True
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : New worker count detected...restarting rebalancer")
logger.info('New worker count detected...restarting rebalancer')
await asyncio.sleep(20)
def main():
@ -417,7 +414,7 @@ def main():
LocalSettings(key='AR-Workers', value='1').save()
worker_count = 1
try:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer initializing...")
logger.info('Rebalancer initializing...')
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.create_task(update_worker_count())
@ -432,13 +429,13 @@ def main():
unknown_error.stop = datetime.now()
unknown_error.save()
loop.run_until_complete(start_queue(worker_count))
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer successfully exited...sleeping for 20 seconds")
logger.info('Rebalancer successfully exited...sleeping for 20 seconds')
sleep(20)
except Exception as e:
error = str(e)
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer loop error: {error}")
logger.error(f'Rebalancer loop error: {error}')
finally:
print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer loop has been terminated")
logger.info('Rebalancer loop has been terminated')
if __name__ == '__main__':
main()

20
trade.py
View file

@ -19,6 +19,8 @@ from os import environ
environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings'
django.setup()
from gui.models import TradeSales, Payments, PaymentHops, Forwards, Peers
import logging
logger = logging.getLogger('[P2P]')
def is_hex(n):
return len(n) % 2 == 0 and all(c in '0123456789ABCDEFabcdef' for c in n)
@ -667,7 +669,7 @@ def getSecret(stub, sale_type):
outgoing_nodes = Forwards.objects.filter(forward_date__gte=filter_30day).values('chan_id_out').annotate(ppm=Round((Sum('fee')/Sum('amt_out_msat'))*1000000000, output_field=IntegerField()), score=Round((Round(Count('id')/1, output_field=IntegerField())+Round(Sum('amt_out_msat')/100000, output_field=IntegerField()))/10, output_field=IntegerField())).exclude(score=0).order_by('-score', '-ppm')[:5]
secret = json.dumps({"incoming_nodes":list(incoming_nodes.values('chan_id_in', 'score', 'ppm')), "outgoing_nodes":list(outgoing_nodes.values('chan_id_out', 'score', 'ppm'))})
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [P2P] : Error getting secret: {str(e)}")
logger.error(f'Error getting secret: {str(e)}')
secret = None
finally:
return secret
@ -680,7 +682,7 @@ def getSecret(stub, sale_type):
payment_nodes = PaymentHops.objects.filter(payment_hash__in=payments_30day).exclude(node_pubkey=self_pubkey).values('node_pubkey').annotate(ppm=Round((Sum('fee')/Sum('amt'))*1000000, output_field=IntegerField()), score=Round((Round(Count('id')/1, output_field=IntegerField())+Round(Sum('amt')/100000, output_field=IntegerField()))/10, output_field=IntegerField())).exclude(score=0).order_by('-score', 'ppm')[:10]
secret = json.dumps({"payment_nodes": list(payment_nodes.values('node_pubkey', 'score', 'ppm'))})
except Exception as e:
print(f"{datetime.now().strftime('%c')} : [P2P] : Error getting secret: {str(e)}")
logger.error(f'Error getting secret: {str(e)}')
secret = None
finally:
return secret
@ -688,9 +690,9 @@ def getSecret(stub, sale_type):
return None
def serve_trades(stub):
print(f"{datetime.now().strftime('%c')} : [P2P] : Serving trades...")
logger.info('Serving trades...')
for trade in get_trades():
print(f"{datetime.now().strftime('%c')} : [P2P] : Serving trade: {trade.id}")
logger.info(f'Serving trade: {trade.id}')
for response in stub.SubscribeCustomMessages(ln.SubscribeCustomMessagesRequest()):
if response.type == 32768:
from_peer = response.peer
@ -703,7 +705,7 @@ def serve_trades(stub):
if 'type' in request:
req_type = request['type']
if req_type == '8050005': # request a seller to finalize a trade or give all open trades
print(f"{datetime.now().strftime('%c')} : [P2P] : SELLER ACTION", '|', 'ID:', request['id'], '|', 'Records:', request['records'])
logger.info(f'SELLER ACTION | ID: {request["id"]} | Records: {request["records"]}')
select_trade = next((record for record in request['records'] if record['type'] == '0'), None)
request_trade = next((record for record in request['records'] if record['type'] == '1'), None)
if request_trade:
@ -723,7 +725,7 @@ def serve_trades(stub):
else:
secret = trade_details.secret
if not secret:
print(f"{datetime.now().strftime('%c')} : [P2P] : Failed to get secret for:", trade_details.id)
logger.error(f'Failed to get secret for: {trade_details.id}')
continue
signerstub = lnsigner.SignerStub(lnd_connect())
shared_key = signerstub.DeriveSharedKey(lns.SharedKeyRequest(ephemeral_pubkey=from_peer)).shared_key
@ -743,16 +745,16 @@ def serve_trades(stub):
trade_details.save()
stub.SendCustomMessage(ln.SendCustomMessageRequest(peer=from_peer, type=32768, data=bytes.fromhex(trade_data)))
else:
print(f"{datetime.now().strftime('%c')} : [P2P] : Expected request type in message:", request['id'])
logger.error(f'Expected request type in message: {request["id"]}')
if 'response' in msg_response:
request = msg_response['response']
if 'failure' in request and request['failure'] != None:
# failure message returned
print(f"{datetime.now().strftime('%c')} : [P2P] : Failure:", request['failure'])
logger.error(f'Failure: {request["failure"]}')
else:
if len(request['records']) == 0:
# message acknowledgements
print(f"{datetime.now().strftime('%c')} : [P2P] : ACK", '|', 'ID:', request['id'])
logger.info(f'ACK | ID: {request["id"]}')
async def get_open_trades(astub, results):
try: