fix: database connection auto-healing and controller fail-fast monitoring (#448)

controller.py

@@ -1,4 +1,5 @@
-import multiprocessing, sys
+import multiprocessing, sys, time
+from datetime import datetime
 import jobs, rebalancer, htlc_stream, p2p, manage
 import logging
 logger = logging.getLogger('[Controller]')
@@ -7,24 +8,75 @@ def run_task(task):
     task()
 
 def main():
-    tasks = [jobs.main, rebalancer.main, htlc_stream.main, p2p.main]
     logger.info('Starting all LNDg processes...')
 
-    processes = []
-    for task in tasks:
-        process = multiprocessing.Process(target=run_task, name=task.__module__, args=(task,))
-        processes.append(process)
-        process.start()
+    tasks_spec = {
+        'jobs': (run_task, (jobs.main,)),
+        'rebalancer': (run_task, (rebalancer.main,)),
+        'htlc_stream': (run_task, (htlc_stream.main,)),
+        'p2p': (run_task, (p2p.main,))
+    }
 
     if len(sys.argv) > 1:
         sys.argv[0] = 'manage.py'
-        process = multiprocessing.Process(target=manage.main(sys.argv), name='manage.py')
-        processes.append(process)
-        process.start()
+        # Pass sys.argv as args to manage.main to avoid blocking parent execution
+        tasks_spec['manage.py'] = (manage.main, (sys.argv,))
 
-    for process in processes:
-        process.join()
-    logger.info('Stopping all LNDg processes...')
+    running_tasks = {}
+
+    for name, (target, args) in tasks_spec.items():
+        process = multiprocessing.Process(target=target, name=name, args=args)
+        process.start()
+        running_tasks[name] = {
+            'target': target,
+            'args': args,
+            'process': process,
+            'last_started': time.time(),
+            'consecutive_failures': 0,
+            'backoff_until': 0.0
+        }
+
+    try:
+        while True:
+            current_time = time.time()
+            for name, info in running_tasks.items():
+                process = info['process']
+                
+                # Check if the process is currently dead (it was running, but has stopped)
+                if process.pid is not None and not process.is_alive() and info['backoff_until'] == 0.0:
+                    exitcode = process.exitcode
+                    uptime = current_time - info['last_started']
+                    
+                    if uptime < 10.0:
+                        info['consecutive_failures'] += 1
+                    else:
+                        info['consecutive_failures'] = 0
+
+                    backoff_delay = min(2 ** info['consecutive_failures'], 60)
+                    info['backoff_until'] = current_time + backoff_delay
+                    
+                    logger.error(
+                        f"Process {name} died (exitcode: {exitcode}, uptime: {uptime:.1f}s). "
+                        f"Restarting in {backoff_delay}s (consecutive failures: {info['consecutive_failures']})."
+                    )
+                    
+                    # Instantiate new Process object
+                    info['process'] = multiprocessing.Process(target=info['target'], name=name, args=info['args'])
+                
+                # If a process is not running (e.g. it just died or is waiting on backoff), and backoff time has passed, start it.
+                if info['process'].pid is None and current_time >= info['backoff_until']:
+                    logger.info(f"Restarting process {name}...")
+                    info['process'].start()
+                    info['last_started'] = time.time()
+                    info['backoff_until'] = 0.0
+
+            time.sleep(2)
+    except KeyboardInterrupt:
+        logger.info('Controller is stopping...')
+        for name, info in running_tasks.items():
+            if info['process'].is_alive():
+                info['process'].terminate()
 
 if __name__ == '__main__':
     main()
+

htlc_stream.py

@@ -61,6 +61,11 @@ def main():
                         del all_forwards[key]
         except Exception as e:
             logger.error(f'Error while running failed HTLC stream: {str(e)}')
+            try:
+                from django.db import connections
+                connections.close_all()
+            except Exception as db_err:
+                logger.error(f"Error closing database connections: {str(db_err)}")
             sleep(20)
 
 if __name__ == '__main__':

jobs.py

@@ -834,6 +834,11 @@ def main():
             agg_failed_htlcs()
         except Exception as e:
             logger.error(f'Error processing background data: {str(e)}')
+            try:
+                from django.db import connections
+                connections.close_all()
+            except Exception as db_err:
+                logger.error(f"Error closing database connections: {str(db_err)}")
         logger.info('Data execution completed...sleeping for 20 seconds')
         sleep(20)
 

p2p.py

@@ -41,6 +41,10 @@ def main():
                     sleep(2)  # polling interval
         except Exception as e:
             logger.error(f'Error running p2p service: {str(e)}')
+            try:
+                django.db.connections.close_all()
+            except Exception as db_err:
+                logger.error(f"Error closing database connections: {str(db_err)}")
         finally:
             if 'p2p_thread' in locals() and p2p_thread.is_alive():
                 logger.info('Removing any remaining processes...')

rebalancer.py

@@ -16,12 +16,20 @@ from gui.models import Rebalancer, Channels, LocalSettings, Forwards, Autopilot
 import logging
 logger = logging.getLogger('[Rebalancer]')
 
+def close_db_connections():
+    try:
+        from django.db import connections
+        connections.close_all()
+    except Exception as e:
+        logger.error(f"Error closing database connections: {str(e)}")
+
 @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:
         logger.error(f'Error getting outbound cands: {str(e)}')
+        close_db_connections()
 
 @sync_to_async
 def save_record(record):
@@ -29,6 +37,7 @@ def save_record(record):
         record.save()
     except Exception as e:
         logger.error(f'Error saving database record: {str(e)}')
+        close_db_connections()
 
 @sync_to_async
 def inbound_cans_len(inbound_cans):
@@ -36,6 +45,7 @@ def inbound_cans_len(inbound_cans):
         return len(inbound_cans)
     except Exception as e:
         logger.error(f'Error getting inbound cands: {str(e)}')
+        close_db_connections()
 
 @sync_to_async
 def check_and_set_allow_multishards():
@@ -265,6 +275,7 @@ def auto_schedule() -> List[Rebalancer]:
         return to_schedule
     except Exception as e:
         logger.error(f'Error scheduling rebalances: {str(e)}')
+        close_db_connections()
         return to_schedule
 
 @sync_to_async
@@ -317,6 +328,7 @@ def auto_enable():
                         logger.debug('Case 5: Pass')
     except Exception as e:
         logger.error(f'Error during auto channel enabling: {str(e)}')
+        close_db_connections()
 
 @sync_to_async
 def get_pending_rebals():
@@ -325,6 +337,7 @@ def get_pending_rebals():
         return rebalances, len(rebalances)
     except Exception as e:
         logger.error(f'Error getting pending rebalances: {str(e)}')
+        close_db_connections()
 
 async def async_queue_manager(rebalancer_queue):
     global scheduled_rebalances, active_rebalances, shutdown_rebalancer
@@ -434,6 +447,7 @@ def main():
     except Exception as e:
         error = str(e)
         logger.error(f'Rebalancer loop error: {error}')
+        close_db_connections()
     finally:
         logger.info('Rebalancer loop has been terminated')