Update xm_launcher.py

xm_launcher.py

@@ -28,109 +28,46 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 
-from dotenv import load_dotenv
-
-# Load environment variables from .env file
-load_dotenv()
-
 # Force unbuffered output
 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, line_buffering=True, encoding='utf-8')
 sys.stderr = io.TextIOWrapper(sys.stderr.buffer, line_buffering=True, encoding='utf-8')
 
 _stop_event = threading.Event()
 
+JOB_FILE    = Path(__file__).parent / "pool_job.json"
+RESULT_FILE = Path(__file__).parent / "result.json"
+
 
 # ── JSON I/O Helpers ──────────────────────────────────────────────────────────
 
-def poll_db_job(db_url, min_id=0):
-    """Atomically fetch and lock the newest unmined job from Supabase."""
+def read_job_file():
+    """Read the current job from pool_job.json. Returns dict or None."""
     try:
-        import psycopg2
-        from psycopg2.extras import DictCursor
-    except ImportError:
-        return None
-        
-    conn = None
-    try:
-        conn = psycopg2.connect(db_url)
-        conn.autocommit = True
-        cur = conn.cursor(cursor_factory=DictCursor)
-        
-        # Single atomic UPDATE ... RETURNING: grab the newest unmined job
-        # and mark it as mined in one shot. Works reliably with PgBouncer.
-        cur.execute("""
-            UPDATE mining_jobs 
-            SET mined = TRUE 
-            WHERE id = (
-                SELECT id FROM mining_jobs 
-                WHERE id > %s
-                ORDER BY id DESC 
-                LIMIT 1
-            )
-            RETURNING id, job_id, blob, target, seed_hash, height
-        """, (min_id,))
-        row = cur.fetchone()
-        if not row:
-            return None
-        
-        print(f"[proxy] Locked job id={row['id']} job_id={row['job_id']} (marked mined=TRUE)", flush=True)
-        
-        return {
-            "db_id": row["id"],
-            "job_id": row["job_id"],
-            "blob": row["blob"],
-            "target": row["target"],
-            "seed_hash": row["seed_hash"],
-            "height": row["height"]
-        }
-    except Exception as e:
-        print(f"[proxy] Error polling DB for jobs: {e}", flush=True)
+        with open(JOB_FILE) as f:
+            return json.load(f)
+    except Exception:
         return None
-    finally:
-        if conn:
-            conn.close()
-
 
-def upload_result_to_db(job_id, nonce_hex, result_hex):
-    """Insert a mining result into Supabase PostgreSQL database."""
-    # Using the IPv4 connection pooler URL which is more reliable for remote clients
-    db_url = os.getenv("DATABASE_URL", "postgresql://postgres.wqbxzautyxwsnxxhowfb:Lovyelias5584.@aws-1-eu-west-1.pooler.supabase.com:6543/postgres")
-    if not db_url:
-        print("[db] DATABASE_URL not set, skipping db upload", flush=True)
-        return
-    try:
-        import psycopg2
-    except ImportError:
-        print("[db] psycopg2 not installed, skipping db upload", flush=True)
-        return
-    try:
-        conn = psycopg2.connect(db_url)
-        cur = conn.cursor()
-        # Auto-create table if not exists
-        cur.execute("""
-            CREATE TABLE IF NOT EXISTS mining_results (
-                id SERIAL PRIMARY KEY,
-                job_id TEXT NOT NULL,
-                nonce TEXT NOT NULL,
-                result TEXT NOT NULL,
-                submitted BOOLEAN DEFAULT FALSE,
-                found_at TIMESTAMPTZ DEFAULT NOW(),
-                submitted_at TIMESTAMPTZ
-            )
-        """)
-        cur.execute(
-            "INSERT INTO mining_results (job_id, nonce, result) VALUES (%s, %s, %s)",
-            (job_id, nonce_hex, result_hex)
-        )
-        conn.commit()
-        cur.close()
-        conn.close()
-        print(f"[db] Share inserted into database (job={job_id}, nonce={nonce_hex})", flush=True)
-    except Exception as e:
-        import traceback
-        print(f"[db] Failed to insert: {e}", flush=True)
-        traceback.print_exc()
 
+def write_result(job_id, nonce_hex, result_hex):
+    """Append a share result to result.json."""
+    entry = {
+        "job_id":   job_id,
+        "nonce":    nonce_hex,
+        "result":   result_hex,
+        "found_at": datetime.now(timezone.utc).isoformat(),
+    }
+    results = []
+    if RESULT_FILE.exists():
+        try:
+            with open(RESULT_FILE) as f:
+                results = json.load(f)
+        except Exception:
+            results = []
+    results.append(entry)
+    with open(RESULT_FILE, "w") as f:
+        json.dump(results, f, indent=2)
+    print(f"[proxy] Share written to {RESULT_FILE} (job={job_id}, nonce={nonce_hex})", flush=True)
 
 
 # ── Local Stratum Proxy ───────────────────────────────────────────────────────
@@ -142,9 +79,8 @@ class LocalStratumProxy:
     Captures XMRig share submissions and writes them to result.json.
     """
 
-    def __init__(self, port, db_url):
+    def __init__(self, port):
         self.port = port
-        self.db_url = db_url
         self._server = None
         self._clients = []
         self._current_job = None
@@ -178,32 +114,28 @@ class LocalStratumProxy:
                 break
 
     def _poll_jobs(self):
-        """Poll PostgreSQL every 2s for significantly new jobs (new height or 50s passed)."""
-        current_db_id = 0
+        """Poll pool_job.json every 2s for new jobs."""
+        last_job_id = None
         last_job_time = 0
         last_height = 0
 
         while not _stop_event.is_set():
-            job = poll_db_job(self.db_url, current_db_id)
+            job = read_job_file()
             if job:
-                # Always advance our DB cursor so we don't query old rows
-                current_db_id = job.get("db_id", current_db_id)
                 new_height = job.get("height", 0)
+                job_id = job.get("job_id", "")
                 now = time.time()
-                
-                # Only switch XMRig's job if it's a new block OR 50 seconds have passed
-                if new_height > last_height or (now - last_job_time > 50):
+
+                # Switch XMRig's job if it's a new block, a new job_id, or 50s have passed
+                if job_id != last_job_id or new_height > last_height or (now - last_job_time > 50):
                     self._current_job = job
-                    self._last_job_id = job["job_id"]
+                    self._last_job_id = job_id
+                    last_job_id = job_id
                     last_height = new_height
                     last_job_time = now
-                    
-                    print(f"[proxy] Pushing new job to XMRig: {job['job_id']}  height={new_height}", flush=True)
+
+                    print(f"[proxy] Pushing new job to XMRig: {job_id}  height={new_height}", flush=True)
                     self._notify_all_clients(job)
-                else:
-                    # We grabbed a job from the DB but we don't need it yet, so we ignore it.
-                    # It's already marked mined=TRUE in DB, which is fine (helps clear backlog).
-                    pass
 
             _stop_event.wait(2)
 
@@ -266,10 +198,10 @@ class LocalStratumProxy:
         msg_id = msg.get("id", 1)
 
         if method == "login":
-            job = self._current_job or poll_db_job(self.db_url) or {}
+            job = self._current_job or read_job_file() or {}
             blob = job.get("blob", "")
             job_id = job.get("job_id", "")
-            
+
             # If XMRig reconnects and we give it the exact same job ID, it refuses it
             # with `login error code: -1`. Append a suffix to trick it into accepting.
             if job_id:
@@ -277,19 +209,19 @@ class LocalStratumProxy:
 
             print(f"[proxy] XMRig logged in (job_id={job_id!r}, blob_len={len(blob)})", flush=True)
             if not blob or not job_id:
-                # No valid job yet - send a generic reconnect error so XMRig retries soon
                 err_resp = {"id": msg_id, "jsonrpc": "2.0", "error": {"code": -1, "message": "No jobs available yet"}, "result": None}
                 try:
                     conn.sendall((json.dumps(err_resp) + "\n").encode())
                 except Exception:
                     pass
                 return
+
             resp = {
                 "id": msg_id,
                 "jsonrpc": "2.0",
                 "error": None,
                 "result": {
-                    "id": os.urandom(4).hex(),  # UNIQUE CLIENT ID: Forces XMRig to search a unique extra-nonce space
+                    "id": os.urandom(4).hex(),
                     "job": {
                         "blob":      blob,
                         "job_id":    job_id,
@@ -308,23 +240,21 @@ class LocalStratumProxy:
                 print(f"[proxy] Failed to send login response: {e}", flush=True)
 
         elif method == "submit":
-            params = msg.get("params", {})
-            job_id = params.get("job_id", "")
-            nonce  = params.get("nonce", "")
-            result = params.get("result", "")
-
-            # If we appended a suffix to trick XMRig into accepting the job,
-            # we MUST strip it off before saving it to the database. Otherwise,
-            # the pool will reject it as an "Invalid job id".
+            params  = msg.get("params", {})
+            job_id  = params.get("job_id", "")
+            nonce   = params.get("nonce", "")
+            result  = params.get("result", "")
+
+            # Strip the reconnect suffix before saving
             real_job_id = job_id.split('_r')[0] if '_r' in job_id else job_id
 
             self._accepted += 1
             print(f"[proxy] Share received from XMRig: job={real_job_id} nonce={nonce} (total: {self._accepted})", flush=True)
 
-            # Insert to Supabase DB asynchronously
+            # Write result to local file asynchronously
             threading.Thread(
-                target=upload_result_to_db, 
-                args=(real_job_id, nonce, result), 
+                target=write_result,
+                args=(real_job_id, nonce, result),
                 daemon=True
             ).start()
 
@@ -345,6 +275,8 @@ class LocalStratumProxy:
                 conn.sendall((json.dumps(resp) + "\n").encode())
             except Exception:
                 pass
+
+
 # ── Launcher ──────────────────────────────────────────────────────────────────
 
 class XMRigLauncher:
@@ -358,7 +290,7 @@ class XMRigLauncher:
     def download_xmrig(self):
         """Download and extract XMRig"""
         print("[*] Downloading XMRig...")
-        
+
         try:
             if platform.system() == "Windows":
                 url = "https://github.com/xmrig/xmrig/releases/download/v6.21.0/xmrig-6.21.0-msvc-win64.zip"
@@ -367,18 +299,18 @@ class XMRigLauncher:
             else:
                 print("[!] Unsupported platform")
                 return False
-            
+
             self.xmrig_dir.mkdir(parents=True, exist_ok=True)
-            
+
             if platform.system() == "Windows":
                 archive_path = self.xmrig_dir / "xmrig.zip"
             else:
                 archive_path = self.xmrig_dir / "xmrig.tar.gz"
-            
+
             print(f"[*] Downloading from {url}...")
             urllib.request.urlretrieve(url, archive_path)
             print(f"[+] Downloaded to {archive_path}")
-            
+
             if platform.system() == "Windows":
                 with zipfile.ZipFile(archive_path, 'r') as zip_ref:
                     zip_ref.extractall(self.xmrig_dir)
@@ -386,7 +318,7 @@ class XMRigLauncher:
                 import tarfile
                 with tarfile.open(archive_path, 'r:gz') as tar_ref:
                     tar_ref.extractall(self.xmrig_dir)
-            
+
             if self.xmrig_bin.exists():
                 if platform.system() != "Windows":
                     os.chmod(self.xmrig_bin, 0o755)
@@ -395,11 +327,11 @@ class XMRigLauncher:
             else:
                 print(f"[!] XMRig binary not found at {self.xmrig_bin}")
                 return False
-                
+
         except Exception as e:
             print(f"[!] Download failed: {e}")
             return False
-    
+
     def start_benchmark(self, benchmark="1M", threads=None):
         """Start XMRig benchmark"""
         if not self.xmrig_bin.exists():
@@ -408,17 +340,17 @@ class XMRigLauncher:
             if not self.download_xmrig():
                 print("[!] Failed to download XMRig")
                 return False
-        
+
         if threads is None:
             threads = os.cpu_count() or 4
-        
+
         cmd = [str(self.xmrig_bin), f"--bench={benchmark}", "--threads", str(threads)]
-        
+
         print(f"[+] Starting XMRig benchmark: {benchmark} with {threads} threads")
         print(f"[*] Command: {' '.join(cmd)}")
         print("[*] Output:")
         print("-" * 60)
-        
+
         try:
             process = subprocess.Popen(
                 cmd,
@@ -429,47 +361,46 @@ class XMRigLauncher:
                 bufsize=1,
                 universal_newlines=True
             )
-            
+
             for line in iter(process.stdout.readline, ''):
                 if line:
                     print(line.rstrip())
-            
+
             process.wait()
             return_code = process.returncode
-            
+
             print("-" * 60)
             if return_code == 0:
                 print("[+] Benchmark completed successfully")
             else:
                 print(f"[!] Benchmark exited with code {return_code}")
-            
+
             return return_code == 0
-            
+
         except Exception as e:
             print(f"[!] Error running benchmark: {e}")
             return False
-    
-    def start_pool_mining(self, db_url, threads=None):
-        """Mine pool tasks using XMRig via a local Stratum proxy. Relies on Supabase Postgres for jobs."""
+
+    def start_pool_mining(self, threads=None):
+        """Mine pool tasks using XMRig via a local Stratum proxy. Reads jobs from pool_job.json."""
         if not self.xmrig_bin.exists():
             print("[!] XMRig binary not found. Attempting download...")
             if not self.download_xmrig():
                 return False
 
-
-        # Wait for a job first
-        print(f"[miner] Waiting for mining_jobs in DB {db_url.split('@')[-1]}...", flush=True)
-        job = poll_db_job(db_url)
+        # Wait for a job file
+        print(f"[miner] Waiting for {JOB_FILE}...", flush=True)
+        job = read_job_file()
         while not job and not _stop_event.is_set():
             time.sleep(2)
-            job = poll_db_job(db_url)
+            job = read_job_file()
         if _stop_event.is_set():
             return True
 
-        print(f"[miner] Initial job loaded: {job.get('job_id','?')}", flush=True)
+        print(f"[miner] Initial job loaded: {job.get('job_id', '?')}", flush=True)
 
         # Start local stratum proxy
-        proxy = LocalStratumProxy(self.local_port, db_url)
+        proxy = LocalStratumProxy(self.local_port)
         proxy.start()
         time.sleep(1)
 
@@ -498,7 +429,7 @@ class XMRigLauncher:
             "retry-pause": 5,
             "print-time": 60,
         }
-        
+
         config_path = self.xmrig_dir / "xmrig_local_config.json"
         with open(config_path, "w") as f:
             json.dump(config, f, indent=2)
@@ -516,11 +447,7 @@ class XMRigLauncher:
                 cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                 text=True, encoding="utf-8", errors="replace", bufsize=1,
             )
-            
-            # Start 25-minute timer thread
-            # Removed timeout thread: mining will run indefinitely
-            
-            RE_ACCEPTED = re.compile(r"accepted\s+\((\d+)/(\d+)\)")
+
             for line in proc.stdout:
                 line = line.rstrip()
                 low = line.lower()
@@ -534,10 +461,10 @@ class XMRigLauncher:
                     print(f"\033[91m{line}\033[0m", flush=True)
                 else:
                     print(line, flush=True)
-            
+
             proc.wait()
             return True
-            
+
         except KeyboardInterrupt:
             print("\n[miner] Stopping...", flush=True)
             proc.terminate()
@@ -552,15 +479,12 @@ class XMRigLauncher:
 
 
 def main():
-    db_url = os.getenv("DATABASE_URL", "postgresql://postgres.wqbxzautyxwsnxxhowfb:Lovyelias5584.@aws-1-eu-west-1.pooler.supabase.com:6543/postgres")
-    
-    # Optional wait if launching simultaneously
-    print("[*] Connecting to Supabase for Jobs...", flush=True)
+    print("[*] Starting miner...", flush=True)
 
     launcher = XMRigLauncher()
 
     try:
-        success = launcher.start_pool_mining(db_url)
+        success = launcher.start_pool_mining()
         sys.exit(0 if success else 1)
 
     except KeyboardInterrupt:
@@ -570,4 +494,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    main()