From 0d65ef3ea6f605cf03c2951dba9041a8deb648ad Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 6 Jul 2026 14:10:13 +0200 Subject: [PATCH] registry event log --- .../contracts/meshnet_contracts/__init__.py | 234 ++++++++++++++---- 1 file changed, 180 insertions(+), 54 deletions(-) diff --git a/packages/contracts/meshnet_contracts/__init__.py b/packages/contracts/meshnet_contracts/__init__.py index f856fc3..1e1f935 100644 --- a/packages/contracts/meshnet_contracts/__init__.py +++ b/packages/contracts/meshnet_contracts/__init__.py @@ -5,21 +5,32 @@ ADR-0007 keeps the public Python boundary stable so real Solana programs can replace this adapter later without changing gateway or tracker call sites. """ -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace import json +import sqlite3 +import threading +import time import urllib.request +import uuid __version__ = "0.1.0" @dataclass(frozen=True) class RegistryWallet: - """Stake, strike, and ban state for a node operator wallet.""" + """Stake, strike, ban, and reputation state for a node operator wallet. + + ``reputation`` is the graduated multiplier of ADR-0018 §6 (scoring deltas + land in issue 08); probation is derived from ``completed_job_count`` vs the + configured probationary job count, so it persists with the wallet record. + """ stake_balance: int = 0 strike_count: int = 0 banned: bool = False completed_job_count: int = 0 + reputation: float = 1.0 + last_audit_ts: float | None = None @dataclass(frozen=True) @@ -100,39 +111,174 @@ class _LocalContractState: probationary_job_count: int = 50 -class RegistryContract: - """Registry wrapper for node stake, strikes, and bans.""" +class RegistryEventLog: + """Event-sourced registry mutations with SQLite persistence and replication. - def __init__(self, state: _LocalContractState, cluster: str) -> None: + Same model as the tracker's ``BillingLedger`` / ``AccountStore``: mutations + are append-only events with unique ids, replayed deterministically on load + and merged idempotently from peers via ``events_since`` / ``apply_events`` + (ADR-0016 §4; the deterministic apply keeps it Raft-compatible, ADR-0019). + """ + + def __init__(self, state: _LocalContractState, db_path: str | None = None) -> None: + self._state = state + self._db_path = db_path + self._lock = threading.RLock() + self._seen_event_ids: set[str] = set() + self._event_log: list[dict] = [] + self._dirty = False + if db_path: + self._init_db() + self._load_from_db() + + def record(self, event_type: str, wallet: str, **fields) -> RegistryWallet: + """Append a locally-originated event and return the resulting wallet.""" + event = { + "id": f"{event_type}-{uuid.uuid4().hex}", + "type": event_type, + "wallet": wallet, + "ts": time.time(), + **fields, + } + with self._lock: + self._apply_locked(event) + return self._state.registry.get(wallet, RegistryWallet()) + + # ---- replication (same model as BillingLedger / AccountStore) ---- + + def events_since(self, index: int) -> tuple[list[dict], int]: + with self._lock: + return list(self._event_log[index:]), len(self._event_log) + + def apply_events(self, events: list[dict]) -> int: + applied = 0 + with self._lock: + for event in events: + event_id = event.get("id") + if not event_id or event_id in self._seen_event_ids: + continue + self._apply_locked(event) + applied += 1 + return applied + + def _apply_locked(self, event: dict) -> None: + etype = event.get("type") + wallet_address = event.get("wallet") + if not wallet_address: + return + current = self._state.registry.get(wallet_address, RegistryWallet()) + if etype == "stake": + updated = replace(current, stake_balance=current.stake_balance + int(event["amount"])) + elif etype == "strike": + strike_count = current.strike_count + 1 + updated = replace( + current, + strike_count=strike_count, + banned=current.banned or strike_count >= int(event.get("ban_threshold", 3)), + ) + elif etype == "slash": + strike_count = current.strike_count + 1 + updated = replace( + current, + stake_balance=max(0, current.stake_balance - int(event["slash_amount"])), + strike_count=strike_count, + banned=current.banned or strike_count >= int(event.get("strike_threshold", 3)), + ) + elif etype == "ban": + updated = replace(current, banned=True) + elif etype == "completed_jobs": + updated = replace( + current, + completed_job_count=current.completed_job_count + int(event.get("count", 1)), + ) + elif etype == "set_reputation": + updated = replace(current, reputation=float(event["reputation"])) + elif etype == "record_audit": + updated = replace(current, last_audit_ts=float(event["audit_ts"])) + else: + return + self._state.registry[wallet_address] = updated + self._seen_event_ids.add(event["id"]) + self._event_log.append(event) + self._dirty = True + + # ---- persistence ---- + + def _init_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.execute( + "CREATE TABLE IF NOT EXISTS registry_events " + "(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)" + ) + con.commit() + con.close() + + def _load_from_db(self) -> None: + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + rows = con.execute( + "SELECT payload FROM registry_events ORDER BY ts, event_id" + ).fetchall() + con.close() + with self._lock: + for (payload,) in rows: + try: + event = json.loads(payload) + except json.JSONDecodeError: + continue + if event.get("id") not in self._seen_event_ids: + self._apply_locked(event) + self._dirty = False + + def save_to_db(self) -> None: + if not self._db_path: + return + with self._lock: + if not self._dirty: + return + events = list(self._event_log) + self._dirty = False + con = sqlite3.connect(self._db_path) # type: ignore[arg-type] + con.executemany( + "INSERT OR IGNORE INTO registry_events (event_id, payload, ts) VALUES (?, ?, ?)", + [(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events], + ) + con.commit() + con.close() + + +class RegistryContract: + """Registry wrapper for node stake, strikes, bans, and reputation.""" + + def __init__(self, state: _LocalContractState, cluster: str, log: RegistryEventLog | None = None) -> None: self._state = state self._cluster = cluster + self._log = log if log is not None else RegistryEventLog(state) def submit_stake(self, wallet_address: str, amount: int) -> dict: if amount <= 0: raise ValueError("stake amount must be positive") - current = self.get_wallet(wallet_address) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance + amount, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count, - ) + self._log.record("stake", wallet_address, amount=amount) return {"signature": f"local-stake-{wallet_address}", "cluster": self._cluster} def get_wallet(self, wallet_address: str) -> RegistryWallet: return self._state.registry.get(wallet_address, RegistryWallet()) def record_strike(self, wallet_address: str, ban_threshold: int = 3) -> dict: - current = self.get_wallet(wallet_address) - strike_count = current.strike_count + 1 - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=strike_count, - banned=current.banned or strike_count >= ban_threshold, - completed_job_count=current.completed_job_count, - ) + self._log.record("strike", wallet_address, ban_threshold=ban_threshold) return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} + def set_reputation(self, wallet_address: str, reputation: float) -> RegistryWallet: + """Persist a graduated reputation score (scoring deltas: issue 08).""" + if reputation < 0: + raise ValueError("reputation must be non-negative") + return self._log.record("set_reputation", wallet_address, reputation=float(reputation)) + + def record_audit(self, wallet_address: str, *, ts: float | None = None) -> RegistryWallet: + """Persist when a wallet was last audited (adaptive audit rate: issue 09).""" + return self._log.record( + "record_audit", wallet_address, audit_ts=float(ts if ts is not None else time.time()) + ) + def submit_slash_proof( self, *, @@ -150,14 +296,12 @@ class RegistryContract: raise ValueError("strike_threshold must be positive") current = self.get_wallet(wallet_address) - stake_after = max(0, current.stake_balance - slash_amount) - strike_count = current.strike_count + 1 - banned = current.banned or strike_count >= strike_threshold - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=stake_after, - strike_count=strike_count, - banned=banned, - completed_job_count=current.completed_job_count, + updated = self._log.record( + "slash", + wallet_address, + slash_amount=slash_amount, + strike_threshold=strike_threshold, + reason=reason, ) receipt = SlashReceipt( signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}", @@ -165,9 +309,9 @@ class RegistryContract: wallet_address=wallet_address, slash_amount=slash_amount, stake_before=current.stake_balance, - stake_after=stake_after, - strike_count=strike_count, - banned=banned, + stake_after=updated.stake_balance, + strike_count=updated.strike_count, + banned=updated.banned, reason=reason, ) self._state.slash_receipts.append(receipt) @@ -175,13 +319,7 @@ class RegistryContract: return receipt def ban_wallet(self, wallet_address: str) -> dict: - current = self.get_wallet(wallet_address) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=True, - completed_job_count=current.completed_job_count, - ) + self._log.record("ban", wallet_address) return {"signature": f"local-ban-{wallet_address}", "cluster": self._cluster} def has_minimum_stake(self, wallet_address: str | None, minimum_stake: int) -> bool: @@ -195,15 +333,7 @@ class RegistryContract: return dict(self._state.registry) def record_completed_job(self, wallet_address: str) -> RegistryWallet: - current = self.get_wallet(wallet_address) - updated = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count + 1, - ) - self._state.registry[wallet_address] = updated - return updated + return self._log.record("completed_jobs", wallet_address, count=1) def probationary_jobs_remaining(self, wallet_address: str) -> int: wallet = self.get_wallet(wallet_address) @@ -331,10 +461,12 @@ class SettlementContract: state: _LocalContractState, cluster: str, cost_per_layer_token_lamport: int, + log: RegistryEventLog | None = None, ) -> None: self._state = state self._cluster = cluster self._cost_per_layer_token_lamport = cost_per_layer_token_lamport + self._log = log if log is not None else RegistryEventLog(state) def settle_epoch( self, @@ -423,13 +555,7 @@ class SettlementContract: return self._state.token_balances.get(wallet_address, 0) def _record_completed_jobs(self, wallet_address: str, count: int) -> None: - current = self._state.registry.get(wallet_address, RegistryWallet()) - self._state.registry[wallet_address] = RegistryWallet( - stake_balance=current.stake_balance, - strike_count=current.strike_count, - banned=current.banned, - completed_job_count=current.completed_job_count + count, - ) + self._log.record("completed_jobs", wallet_address, count=count) def deployment_plan(self) -> dict: """Return the configured manual testnet deployment targets."""