Add SQLite persistence to Hub for safe reboots

.gitignore

@@ -0,0 +1 @@
+\n*.db\n*.sqlite3

hub/db.py

@@ -0,0 +1,71 @@
+import sqlite3
+import os
+from typing import Optional
+
+DB_FILE = "ledger.db"
+
+def init_db():
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS ledger (
+            node_id TEXT PRIMARY KEY,
+            balance REAL NOT NULL
+        )
+    ''')
+    conn.commit()
+    conn.close()
+
+def get_balance(node_id: str) -> Optional[float]:
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    cursor.execute("SELECT balance FROM ledger WHERE node_id = ?", (node_id,))
+    row = cursor.fetchone()
+    conn.close()
+    if row:
+        return row[0]
+    return None
+
+def set_balance(node_id: str, balance: float):
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    cursor.execute('''
+        INSERT INTO ledger (node_id, balance)
+        VALUES (?, ?)
+        ON CONFLICT(node_id) DO UPDATE SET balance = excluded.balance
+    ''', (node_id, balance))
+    conn.commit()
+    conn.close()
+
+def add_balance(node_id: str, amount: float):
+    # Atomic addition
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    # Ensure node exists first
+    cursor.execute("SELECT balance FROM ledger WHERE node_id = ?", (node_id,))
+    row = cursor.fetchone()
+    if row is None:
+        # Should not happen if correctly initialized, but fallback to amount
+        cursor.execute("INSERT INTO ledger (node_id, balance) VALUES (?, ?)", (node_id, amount))
+    else:
+        cursor.execute("UPDATE ledger SET balance = balance + ? WHERE node_id = ?", (amount, node_id))
+    conn.commit()
+    conn.close()
+
+def deduct_balance(node_id: str, amount: float) -> bool:
+    # Atomic deduction, returns True if successful, False if insufficient balance
+    conn = sqlite3.connect(DB_FILE)
+    cursor = conn.cursor()
+    cursor.execute("SELECT balance FROM ledger WHERE node_id = ?", (node_id,))
+    row = cursor.fetchone()
+    if row is None or row[0] < amount:
+        conn.close()
+        return False
+        
+    cursor.execute("UPDATE ledger SET balance = balance - ? WHERE node_id = ?", (amount, node_id))
+    conn.commit()
+    conn.close()
+    return True
+
+# Initialize database on module import
+init_db()

hub/main.py

@@ -1,42 +1,48 @@
 from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
 from typing import Dict, List
 import uuid
+import db
 
 from models import NodeRegistration, TaskCreate, TaskResult, NodeBalance, TaskBid
 
 app = FastAPI(title="Chronos Protocol L1 Hub", description="The Time Exchange Clearinghouse", version="0.1.1")
 
-# In-memory storage for MVP
-ledger: Dict[str, float] = {}  # node_id -> balance
+# In-memory storage for active tasks
 active_tasks: Dict[str, dict] = {} # task_id -> task_details
 completed_tasks: Dict[str, dict] = {} # task_id -> result
 connected_nodes: Dict[str, WebSocket] = {} # node_id -> websocket
 
 @app.post("/register")
 async def register_node(node: NodeRegistration):
-    if node.pubkey not in ledger:
-        ledger[node.pubkey] = 10.0 # Starter bonus
-    return {"status": "success", "node_id": node.pubkey, "balance": ledger[node.pubkey]}
+    balance = db.get_balance(node.pubkey)
+    if balance is None:
+        db.set_balance(node.pubkey, 10.0) # Starter bonus
+        balance = 10.0
+    return {"status": "success", "node_id": node.pubkey, "balance": balance}
 
 @app.get("/balance/{node_id}")
 async def get_balance(node_id: str):
-    if node_id not in ledger:
+    balance = db.get_balance(node_id)
+    if balance is None:
         raise HTTPException(status_code=404, detail="Node not found")
-    return {"node_id": node_id, "balance_seconds": ledger[node_id]}
+    return {"node_id": node_id, "balance_seconds": balance}
 
 @app.post("/tasks/submit")
 async def submit_task(task: TaskCreate):
-    if task.consumer_id not in ledger:
+    consumer_balance = db.get_balance(task.consumer_id)
+    if consumer_balance is None:
         raise HTTPException(status_code=404, detail="Consumer node not found")
         
     # If bounty is positive, consumer is PAYING. Check consumer balance.
-    if task.bounty > 0 and ledger[task.consumer_id] < task.bounty:
+    if task.bounty > 0 and consumer_balance < task.bounty:
         raise HTTPException(status_code=400, detail="Insufficient SECONDS balance to pay for task")
         
     # Note: If bounty is negative, consumer is SELLING data. We don't deduct here.
     # We will deduct from the provider when they complete the task.
     if task.bounty > 0:
-        ledger[task.consumer_id] -= task.bounty
+        success = db.deduct_balance(task.consumer_id, task.bounty)
+        if not success:
+            raise HTTPException(status_code=400, detail="Insufficient SECONDS balance")
         
     task_id = str(uuid.uuid4())
     task_data = {
@@ -106,21 +112,23 @@ async def complete_task(result: TaskResult):
     if not task:
         raise HTTPException(status_code=404, detail="Task not found or already claimed")
         
-    if result.provider_id not in ledger:
-        ledger[result.provider_id] = 0.0
+    provider_balance = db.get_balance(result.provider_id)
+    if provider_balance is None:
+        db.set_balance(result.provider_id, 0.0)
 
     # Transfer SECONDS based on positive or negative bounty
     bounty = task["bounty"]
     if bounty >= 0:
         # Standard Compute Market: Provider earns SECONDS
-        ledger[result.provider_id] += bounty
+        db.add_balance(result.provider_id, bounty)
     else:
         # Data Market: Provider PAYS to receive this payload/task
         cost = abs(bounty)
-        if ledger[result.provider_id] < cost:
+        success = db.deduct_balance(result.provider_id, cost)
+        if not success:
             raise HTTPException(status_code=400, detail="Provider lacks SECONDS to buy this data")
-        ledger[result.provider_id] -= cost
-        ledger[task["consumer_id"]] += cost # The sender earns SECONDS
+        
+        db.add_balance(task["consumer_id"], cost) # The sender earns SECONDS
     
     # Move task to completed
     task["status"] = "completed"
@@ -145,7 +153,7 @@ async def complete_task(result: TaskResult):
         except:
             pass # Consumer disconnected, they can fetch it via REST later (TODO)
 
-    return {"status": "success", "earned": task["bounty"], "new_balance": ledger[result.provider_id]}
+    return {"status": "success", "earned": task["bounty"], "new_balance": db.get_balance(result.provider_id)}
 
 @app.websocket("/ws/{node_id}")
 async def websocket_endpoint(websocket: WebSocket, node_id: str):