unclaimed.py

#!/usr/bin/env python3
"""Find all participants whose reward was assigned by SettleAccounts but
SKIPPED at the SetSettleAmount write (the "negative coin balance" /
ErrNegativeCoinBalance settle bug).

On-chain signature, both conditions required:

  1. epoch_performance_summary[N][addr].rewarded_coins > 0
     (the chain computed a positive RewardCoins for this participant)

  2. settle_amount snapshot at height = epoch_meta[N+1].effective_block_height
     does NOT contain (epoch=N, participant=addr)
     (the chain skipped writing SetSettleAmount, so the participant could
     never claim — settleError such as ErrNegativeCoinBalance caused the
     `continue` in the second loop of SettleAccounts)

Pairs satisfying both are written to the output CSV with per-epoch breakdown
and per-address total. Pairs failing condition 2 (SettleAmount existed)
are dropped — those are the participant's own responsibility (missed claim,
bad signature, missed validations, etc.).

Output schema:

    address, e1, e2, ..., eN, total_ngonka, total_gnk

Usage:
    python3 unclaimed.py NODE_IP [--from-epoch N] [--to-epoch N] [--out PATH]

Requires an archive full node — historical state (epoch_performance_summary
for old epochs, settle_amount snapshot at past heights) is needed.
"""
import argparse
import asyncio
import aiohttp
import json
import os
import sqlite3
import sys
import time
import urllib.parse


CONCURRENCY = 40
CACHE_SCHEMA_VERSION = 2  # bumped: now also caches settle_amount snapshots


async def get_json(session, url, retries=8, headers=None):
    last_err = None
    for attempt in range(retries):
        try:
            async with session.get(url, headers=headers,
                                   timeout=aiohttp.ClientTimeout(total=60)) as r:
                if r.status == 200:
                    return await r.json()
                if r.status == 404:
                    return {"_error": "404"}
                if r.status == 429:
                    await asyncio.sleep(2 + attempt * 2)
                    continue
                if attempt == retries - 1:
                    body = await r.text()
                    return {"_error": f"HTTP {r.status}: {body[:200]}"}
        except Exception as e:
            last_err = e
            await asyncio.sleep(1 + attempt)
    return {"_error": f"retries exhausted: {last_err}"}


async def fetch_current_epoch(session, base):
    d = await get_json(session, f"{base}/chain-api/productscience/inference/inference/get_current_epoch")
    return int(d["epoch"])


async def fetch_epoch_group(session, base, epoch):
    return await get_json(session, f"{base}/chain-api/productscience/inference/inference/epoch_group_data/{epoch}")


async def fetch_epoch_summary(session, base, epoch, addr, sem):
    async with sem:
        return await get_json(session, f"{base}/chain-api/productscience/inference/inference/epoch_performance_summary/{epoch}/{addr}")


async def fetch_settle_snapshot_at_height(session, base, height):
    """Return the full settle_amount table at a given historical block,
    paginated to completion. Each entry is {participant, epoch_index, ...}."""
    out = []
    next_key = None
    headers = {"x-cosmos-block-height": str(height)}
    while True:
        params = ["pagination.limit=500"]
        if next_key:
            params.append("pagination.key=" + urllib.parse.quote(next_key))
        url = (f"{base}/chain-api/productscience/inference/inference/settle_amount"
               f"?{'&'.join(params)}")
        d = await get_json(session, url, headers=headers)
        if "_error" in d:
            raise SystemExit(f"settle_amount snapshot at h={height} failed: {d['_error']}")
        out.extend(d.get("settle_amount", []) or [])
        next_key = (d.get("pagination") or {}).get("next_key")
        if not next_key:
            break
    return out


def open_cache(path):
    db = sqlite3.connect(path)
    db.executescript("""
        CREATE TABLE IF NOT EXISTS schema_meta(key TEXT PRIMARY KEY, value TEXT);
        CREATE TABLE IF NOT EXISTS epoch_meta(
            epoch INTEGER PRIMARY KEY,
            eff INTEGER,
            members TEXT
        );
        CREATE TABLE IF NOT EXISTS rewards(
            epoch INTEGER,
            addr TEXT,
            rewarded_coins TEXT,
            claimed INTEGER,
            PRIMARY KEY(epoch, addr)
        );
        CREATE TABLE IF NOT EXISTS settle_present(
            epoch INTEGER,
            addr TEXT,
            PRIMARY KEY(epoch, addr)
        );
        CREATE TABLE IF NOT EXISTS settle_snapshot_done(
            epoch INTEGER PRIMARY KEY
        );
    """)
    row = db.execute('SELECT value FROM schema_meta WHERE key="version"').fetchone()
    cur_version = int(row[0]) if row else 0
    if cur_version < CACHE_SCHEMA_VERSION:
        # Schema changed (eff column added to epoch_meta + new settle tables).
        # Wipe affected tables to force re-fetch, keep rewards cache.
        db.execute("DROP TABLE IF EXISTS epoch_meta")
        db.executescript("""
            CREATE TABLE IF NOT EXISTS epoch_meta(
                epoch INTEGER PRIMARY KEY,
                eff INTEGER,
                members TEXT
            );
        """)
        db.execute("INSERT OR REPLACE INTO schema_meta VALUES ('version', ?)",
                   (str(CACHE_SCHEMA_VERSION),))
        print(f"[cache] schema upgraded to v{CACHE_SCHEMA_VERSION}", file=sys.stderr)
    db.commit()
    return db


async def main_async(node_ip, from_epoch, to_epoch, out_path):
    base = node_ip if node_ip.startswith("http") else f"http://{node_ip}:8000"
    here = os.path.dirname(os.path.abspath(__file__))
    cache_dir = os.path.join(here, f'cache_unclaimed_{node_ip.replace("/", "_").replace(":", "_")}')
    os.makedirs(cache_dir, exist_ok=True)
    db_path = os.path.join(cache_dir, "cache.db")
    db = open_cache(db_path)
    print(f"cache: {db_path}", file=sys.stderr)

    connector = aiohttp.TCPConnector(limit=CONCURRENCY)
    async with aiohttp.ClientSession(connector=connector) as session:
        cur_epoch = await fetch_current_epoch(session, base)
        if to_epoch is None:
            to_epoch = cur_epoch - 2
        print(f"node : {base}", file=sys.stderr)
        print(f"epoch range : [{from_epoch}..{to_epoch}] (chain at {cur_epoch})", file=sys.stderr)

        sem = asyncio.Semaphore(CONCURRENCY)
        # We also need epoch_meta for (to_epoch + 1) to know the snapshot height
        # for to_epoch's SettleAmount check.
        epochs = list(range(from_epoch, to_epoch + 1))
        epochs_meta_needed = list(range(from_epoch, to_epoch + 2))

        # 1. Fetch epoch_group_data for each epoch (members + effective_block_height)
        cached_meta = {e for (e,) in db.execute("SELECT epoch FROM epoch_meta").fetchall()}
        todo_meta = [e for e in epochs_meta_needed if e not in cached_meta]
        print(f"epoch_meta: {len(epochs_meta_needed)} target, {len(todo_meta)} to fetch", file=sys.stderr)
        for i in range(0, len(todo_meta), CONCURRENCY):
            chunk = todo_meta[i:i + CONCURRENCY]
            results = await asyncio.gather(*[fetch_epoch_group(session, base, e) for e in chunk])
            rows = []
            for e, d in zip(chunk, results):
                egd = d.get("epoch_group_data")
                if not egd:
                    print(f"  epoch {e}: no data ({d.get('_error', '')})", file=sys.stderr)
                    continue
                eff = int(egd.get("effective_block_height", 0) or 0)
                members = [m["member_address"] for m in (egd.get("validation_weights") or [])]
                rows.append((e, eff, json.dumps(members)))
            db.executemany("INSERT OR REPLACE INTO epoch_meta VALUES (?, ?, ?)", rows)
            db.commit()
            print(f"  epoch_meta {i + len(chunk)}/{len(todo_meta)}", file=sys.stderr)

        # 2. Fetch per-participant rewards (epoch_performance_summary)
        for e in epochs:
            row = db.execute("SELECT members FROM epoch_meta WHERE epoch=?", (e,)).fetchone()
            if not row:
                print(f"  epoch {e}: no meta in cache, skipping", file=sys.stderr)
                continue
            members = json.loads(row[0])
            cached_addrs = {a for (a,) in db.execute(
                "SELECT addr FROM rewards WHERE epoch=?", (e,)
            ).fetchall()}
            need = [a for a in members if a not in cached_addrs]
            if not need:
                continue
            t0 = time.time()
            results = await asyncio.gather(*[fetch_epoch_summary(session, base, e, a, sem) for a in need])
            rows = []
            for addr, resp in zip(need, results):
                p = (resp or {}).get("epochPerformanceSummary", {})
                if not p:
                    rows.append((e, addr, "0", 0))
                    continue
                rows.append((
                    e, addr,
                    p.get("rewarded_coins", "0"),
                    1 if p.get("claimed") else 0,
                ))
            db.executemany("INSERT OR REPLACE INTO rewards VALUES (?, ?, ?, ?)", rows)
            db.commit()
            print(f"  epoch {e}: {len(rows)}/{len(need)} rewards ({time.time()-t0:.1f}s)", file=sys.stderr)

        # 3. Fetch settle_amount snapshot at eff_{N+1} for every epoch we care about
        snapshot_done = {e for (e,) in db.execute(
            "SELECT epoch FROM settle_snapshot_done"
        ).fetchall()}
        todo_snap = [e for e in epochs if e not in snapshot_done]
        print(f"settle snapshots: {len(epochs)} target, {len(todo_snap)} to fetch",
              file=sys.stderr)
        for e in todo_snap:
            # Get height = effective_block_height of epoch (e+1)
            row = db.execute("SELECT eff FROM epoch_meta WHERE epoch=?", (e + 1,)).fetchone()
            if not row or not row[0]:
                print(f"  epoch {e}: missing eff for epoch {e+1}, skipping snapshot",
                      file=sys.stderr)
                continue
            height = int(row[0])
            t0 = time.time()
            entries = await fetch_settle_snapshot_at_height(session, base, height)
            rows = []
            count_this_epoch = 0
            for entry in entries:
                e_idx = int(entry.get("epoch_index", 0) or 0)
                p = entry.get("participant", "")
                if p:
                    rows.append((e_idx, p))
                    if e_idx == e:
                        count_this_epoch += 1
            db.executemany("INSERT OR IGNORE INTO settle_present VALUES (?, ?)", rows)
            db.execute("INSERT OR REPLACE INTO settle_snapshot_done VALUES (?)", (e,))
            db.commit()
            print(f"  epoch {e}: snapshot h={height} -> {len(entries)} entries, "
                  f"{count_this_epoch} for ep {e} ({time.time()-t0:.1f}s)", file=sys.stderr)

        # 4. Pick the bug victims:
        #    rewarded_coins > 0 AND (epoch, addr) NOT in settle_present
        all_epochs = list(range(from_epoch, to_epoch + 1))
        per_addr_per_epoch = {}
        totals = {}
        cur = db.execute(f"""
            SELECT r.epoch, r.addr, r.rewarded_coins
            FROM rewards r
            WHERE CAST(r.rewarded_coins AS INTEGER) > 0
              AND NOT EXISTS (
                SELECT 1 FROM settle_present sp
                WHERE sp.epoch = r.epoch AND sp.addr = r.addr
              )
              AND r.epoch BETWEEN ? AND ?
              AND r.epoch IN (SELECT epoch FROM settle_snapshot_done)
        """, (from_epoch, to_epoch))
        for ep, addr, rc in cur:
            n = int(rc)
            per_addr_per_epoch.setdefault(addr, {})[ep] = n
            totals[addr] = totals.get(addr, 0) + n

        sorted_pairs = sorted(totals.items(), key=lambda kv: -kv[1])

        with open(out_path, "w") as f:
            header = ["address"] + [str(e) for e in all_epochs] + ["total_ngonka", "total_gnk"]
            f.write(",".join(header) + "\n")
            for addr, total in sorted_pairs:
                shares = per_addr_per_epoch.get(addr, {})
                cols = [addr] + [str(shares.get(e, 0)) for e in all_epochs] + \
                       [str(total), f"{total/1e9:.9f}"]
                f.write(",".join(cols) + "\n")

        print("\nResult (settle-dropped bug victims only):", file=sys.stderr)
        print(f"  affected addresses : {len(totals):,}", file=sys.stderr)
        print(f"  total unclaimed    : {sum(totals.values()):,} ngonka "
              f"({sum(totals.values())/1e9:,.6f} GNK)", file=sys.stderr)
        print(f"  → {out_path}", file=sys.stderr)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("node_ip", help="archive node hostname or IP")
    p.add_argument("--from-epoch", type=int, default=1, help="first epoch (default: 1)")
    p.add_argument("--to-epoch", type=int, default=None, help="last epoch (default: current-2)")
    p.add_argument("--out", default="unclaimed.csv", help="output CSV path")
    args = p.parse_args()
    asyncio.run(main_async(args.node_ip, args.from_epoch, args.to_epoch, args.out))


if __name__ == "__main__":
    main()