Merge branch 'fairminterv2' into protocolchanges

Author: Ouziel

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev:  v0.9.9
+    rev: v0.9.10
     hooks:
     -   id: ruff
         args: [ --fix ]

apiary.apib

@@ -1,4 +1,4 @@
 [run]
-concurrency = multiprocessing
+concurrency = multiprocessing,thread
 parallel = true
 sigterm = true

counterparty-core/.coveragerc

@@ -1946,6 +1946,7 @@ def get_balances_by_addresses(
     state_db,
     addresses: str,
     type: BalanceType = "all",  # pylint: disable=W0622
+    asset: str = None,
     cursor: str = None,
     limit: int = 100,
     offset: int = None,
@@ -1955,32 +1956,36 @@ def get_balances_by_addresses(
     Returns the balances of several addresses
     :param str addresses: Comma separated list of addresses (e.g. $ADDRESS_1,$ADDRESS_2)
     :param str type: The type of balances to return
+    :param str asset: The asset to return (e.g. XCP)
     :param str cursor: The last index of the balances to return
     :param int limit: The maximum number of balances to return (e.g. 5)
     :param int offset: The number of lines to skip before returning results (overrides the `cursor` parameter)
     :param str sort: The sort order of the balances to return (overrides the `cursor` parameter) (e.g. quantity:desc)
     """
-    where = [
-        {"address__in": addresses.split(","), "quantity__gt": 0},
-        {"utxo_address__in": addresses.split(","), "quantity__gt": 0},
-    ]
-    if type == "utxo":
-        where.pop(0)
-    elif type == "address":
-        where.pop(1)
-
-    assets_result = select_rows(
-        state_db,
-        "balances",
-        select="DISTINCT asset AS asset",
-        where=where,
-        order="ASC",
-        cursor_field="asset",
-        last_cursor=cursor,
-        offset=offset,
-        limit=limit,
-    )
-    assets = [asset["asset"] for asset in assets_result.result]
+    if asset is None:
+        where = [
+            {"address__in": addresses.split(","), "quantity__gt": 0},
+            {"utxo_address__in": addresses.split(","), "quantity__gt": 0},
+        ]
+        if type == "utxo":
+            where.pop(0)
+        elif type == "address":
+            where.pop(1)
+
+        assets_result = select_rows(
+            state_db,
+            "balances",
+            select="DISTINCT asset AS asset",
+            where=where,
+            order="ASC",
+            cursor_field="asset",
+            last_cursor=cursor,
+            offset=offset,
+            limit=limit,
+        )
+        assets = [asset["asset"] for asset in assets_result.result]
+    else:
+        assets = [asset]
 
     where = [
         {"address__in": addresses.split(","), "asset__in": assets, "quantity__gt": 0},
@@ -2028,8 +2033,11 @@ def get_balances_by_addresses(
                 }
             )
         result.append(current_balances)
-
-    return QueryResult(result, assets_result.next_cursor, "balances", assets_result.result_count)
+    if asset is None:
+        return QueryResult(
+            result, assets_result.next_cursor, "balances", assets_result.result_count
+        )
+    return QueryResult(result, None, "balances", 1)
 
 
 def get_balances_by_address_and_asset(

counterparty-core/counterpartycore/lib/api/queries.py

@@ -42,7 +42,7 @@ def clean_url_for_log(url):
 
 # for testing
 def should_retry():
-    if CurrentState().ledger_state() == "Stopping":
+    if CurrentState().stopping():
         return False
     return True
 

counterparty-core/counterpartycore/lib/backend/bitcoind.py

@@ -1,11 +1,9 @@
 import _thread
 import binascii
-import cProfile
 import decimal
 import logging
 import multiprocessing
 import os
-import pstats
 import threading
 import time
 
@@ -20,10 +18,11 @@
 )
 from counterpartycore.lib.api import apiserver as api_v2
 from counterpartycore.lib.api import apiv1, dbbuilder
+from counterpartycore.lib.backend import rsfetcher
 from counterpartycore.lib.cli import bootstrap, log
 from counterpartycore.lib.ledger.backendheight import BackendHeight
 from counterpartycore.lib.ledger.currentstate import CurrentState
-from counterpartycore.lib.monitors import slack
+from counterpartycore.lib.monitors import profiler, slack
 from counterpartycore.lib.parser import blocks, check, follow
 from counterpartycore.lib.utils import database, helpers
 
@@ -87,7 +86,7 @@ def __init__(self, args, log_stream=None, stop_when_ready=False):
         self.api_stop_event = None
         self.backend_height_thread = None
         self.log_stream = log_stream
-        self.profiler = None
+        self.periodic_profiler = None
         self.stop_when_ready = stop_when_ready
         self.stopped = False
 
@@ -174,14 +173,12 @@ def run_server(self):
 
         # Catch Up
         if config.PROFILE:
-            logger.info("Starting profiler before catchup...")
-            self.profiler = cProfile.Profile()
-            self.profiler.enable()
-            blocks.catch_up(self.db)
-            logger.info("Stopping profiler after catchup...")
-            self.profiler.disable()
-        else:
-            blocks.catch_up(self.db)
+            self.periodic_profiler = profiler.PeriodicProfilerThread(
+                interval_minutes=config.PROFILE_INTERVAL_MINUTES
+            )
+            self.periodic_profiler.start()
+
+        blocks.catch_up(self.db)
 
         if self.stop_when_ready:
             slack.trigger_webhook()
@@ -206,8 +203,9 @@ def stop(self):
         if self.stopped:
             return
         logger.info("Shutting down...")
-        if self.db:
-            CurrentState().set_ledger_state(self.db, "Stopping")
+
+        CurrentState().set_stopping()
+        rsfetcher.RSFetcher().stop()
 
         # Ensure all threads are stopped
         if self.follower_daemon:
@@ -225,19 +223,9 @@ def stop(self):
         if self.apiserver_v2:
             self.apiserver_v2.stop()
 
-        # Dump profiling data
-        if config.PROFILE and self.profiler:
-            logger.info("Dumping profiling data...")
-            profile_path = os.path.join(config.CACHE_DIR, "catchup.prof")
-            try:
-                self.profiler.dump_stats(profile_path)
-                stats = pstats.Stats(self.profiler).sort_stats("cumtime")
-                stats.print_stats(20)
-                stats = pstats.Stats(self.profiler).sort_stats("tottime")
-                stats.print_stats(20)
-            except Exception as e:  # pylint: disable=broad-except
-                logger.error("Error dumping profiler stats: %s", e)
-            self.profiler = None
+        if self.periodic_profiler:
+            logger.info("Stopping periodic profiler...")
+            self.periodic_profiler.stop()
 
         self.stopped = True
         logger.info("Shutdown complete.")

counterparty-core/counterpartycore/lib/cli/server.py

@@ -7,7 +7,7 @@
 
 
 # Semantic Version
-__version__ = "10.10.1"  # for hatch
+__version__ = "11.0.0-alpha.1"  # for hatch
 VERSION_STRING = __version__
 version = VERSION_STRING.split("-", maxsplit=1)[0].split(".")
 VERSION_MAJOR = int(version[0])
@@ -29,6 +29,7 @@
         "10.8.0": [("rollback", 871780)],
         "10.9.0-rc.1": [("rollback", 871780)],
         "10.9.0": [("rollback", 871780)],
+        "11.0.0-alpha.1": [("reparse", 0)],
     },
     "testnet3": {
         "10.3.0": [("reparse", 0)],
@@ -39,9 +40,11 @@
         "10.9.0-rc.1": [("rollback", 3522632)],
         "10.9.0": [("rollback", 3522632)],
         "10.10.0": [("rollback", 3522632)],
+        "11.0.0-alpha.1": [("reparse", 0)],
     },
     "testnet4": {
         "10.10.0": [("rollback", 64492)],
+        "11.0.0-alpha.1": [("reparse", 0)],
     },
 }
 
@@ -250,3 +253,5 @@
 CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
 LEDGER_DB_MIGRATIONS_DIR = os.path.join(CURRENT_DIR, "ledger", "migrations")
 STATE_DB_MIGRATIONS_DIR = os.path.join(CURRENT_DIR, "api", "migrations")
+
+PROFILE_INTERVAL_MINUTES = 15

counterparty-core/counterpartycore/lib/config.py

@@ -75,6 +75,12 @@ def ledger_state(self):
         with LedgerDBConnectionPool().connection() as ledger_db:
             return get_config_value(ledger_db, "LEDGER_STATE") or "Starting"
 
+    def set_stopping(self):
+        self.state["STOPPING"] = True
+
+    def stopping(self):
+        return self.state.get("STOPPING", False)
+
 
 class ConsensusHashBuilder(metaclass=helpers.SingletonMeta):
     def __init__(self):

counterparty-core/counterpartycore/lib/ledger/currentstate.py

@@ -269,8 +269,7 @@ def debit(db, address, asset, quantity, tx_index, action=None, event=None):
     }
     insert_record(db, "debits", bindings, "DEBIT")
 
-    ConsensusHashBuilder().append_to_block_ledger(f"{block_index}{address}{asset}{quantity}")
-    ConsensusHashBuilder().append_to_block_migration(
+    ConsensusHashBuilder().append_to_block_ledger(
         f"{block_index}{str(address)[0:36]}{asset}{quantity}"
     )
     return utxo_address
@@ -349,8 +348,7 @@ def credit(db, address, asset, quantity, tx_index, action=None, event=None):
     }
     insert_record(db, "credits", bindings, "CREDIT")
 
-    ConsensusHashBuilder().append_to_block_ledger(f"{block_index}{address}{asset}{quantity}")
-    ConsensusHashBuilder().append_to_block_migration(
+    ConsensusHashBuilder().append_to_block_ledger(
         f"{block_index}{str(address)[0:36]}{asset}{quantity}"
     )