Merge pull request #2308 from CounterpartyXCP/develop

v10.4.3

.github/workflows/test_compose.yml

@@ -2,7 +2,7 @@ name: Docker Compose
 
 on:
   push:
-    branches: ['develop', 'master']
+    branches: ['develop', 'master', 'fixes']
 
 jobs:
   build:

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

@@ -76,7 +76,8 @@ def api_root():
         "backend_height": BACKEND_HEIGHT,
         "counterparty_height": counterparty_height,
         "documentation": "https://counterpartycore.docs.apiary.io/",
-        "blueprint": f"{request.url_root}v2/blueprint",
+        "routes": f"{request.url_root}v2/routes",
+        "blueprint": "https://raw.githubusercontent.com/CounterpartyXCP/counterparty-core/refs/heads/master/apiary.apib",
     }
 
 
@@ -155,6 +156,7 @@ def return_result(
     response = flask.make_response(to_json(api_result), http_code)
     response.headers["X-COUNTERPARTY-HEIGHT"] = util.CURRENT_BLOCK_INDEX
     response.headers["X-COUNTERPARTY-READY"] = is_server_ready()
+    response.headers["X-COUNTERPARTY-VERSION"] = config.VERSION_STRING
     response.headers["X-BITCOIN-HEIGHT"] = BACKEND_HEIGHT
     response.headers["Content-Type"] = "application/json"
     if not config.API_NO_ALLOW_CORS:
@@ -308,6 +310,7 @@ def handle_route(**kwargs):
             result = execute_api_function(db, rule, route, function_args)
     except (
         exceptions.JSONRPCInvalidRequest,
+        flask.wrappers.BadRequest,
         exceptions.TransactionError,
         exceptions.BalanceError,
         exceptions.UnknownPubKeyError,

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

@@ -427,7 +427,8 @@ def adjust_get_transactions_results(query_result):
     """Format the data field.  Try and decode the data from a utf-8 uncoded string. Invalid utf-8 strings return an empty data."""
     filtered_results = []
     for transaction_row in list(query_result):
-        transaction_row["data"] = transaction_row["data"].hex()
+        if isinstance(transaction_row["data"], bytes):
+            transaction_row["data"] = transaction_row["data"].hex()
         filtered_results.append(transaction_row)
     return filtered_results
 

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

@@ -534,9 +534,18 @@ def rollback_assets_info(api_db, event):
 
 
 def execute_event(api_db, event):
+    cursor = api_db.cursor()
     sql, sql_bindings = event_to_sql(event)
     if sql is not None:
-        cursor = api_db.cursor()
+        # check if the transaction is already here from the mempool
+        if event["event"] == "NEW_TRANSACTION":
+            check_tx_sql = "SELECT * FROM transactions WHERE tx_hash = :tx_hash"
+            if fetch_one(
+                api_db,
+                check_tx_sql,
+                {"tx_hash": event["tx_hash"], "block_index": config.MEMPOOL_BLOCK_INDEX},
+            ):
+                clean_mempool(api_db)
         cursor.execute(sql, sql_bindings)
         if event["command"] == "insert":
             cursor.execute("SELECT last_insert_rowid() AS rowid")
@@ -642,7 +651,7 @@ def parse_event(api_db, event, catching_up=False):
         return
     with api_db:
         logger.trace(f"API Watcher - Parsing event: {event}")
-        if event["event"] == "NEW_BLOCK" and not catching_up:
+        if event["event"] == "NEW_BLOCK":
             clean_mempool(api_db)
         event["insert_rowid"] = execute_event(api_db, event)
         update_balances(api_db, event)
@@ -653,7 +662,7 @@ def parse_event(api_db, event, catching_up=False):
         update_fairminters(api_db, event)
         insert_event(api_db, event)
         logger.event(f"API Watcher - Event parsed: {event['message_index']} {event['event']}")
-        if event["event"] == "BLOCK_PARSED" and not catching_up:
+        if event["event"] == "BLOCK_PARSED":
             synchronize_mempool(api_db, api_db)
 
 
@@ -742,15 +751,15 @@ def parse_next_event(api_db, ledger_db):
         next_event_sql = "SELECT * FROM messages ORDER BY message_index ASC LIMIT 1"
         next_event = fetch_one(ledger_db, next_event_sql)
         parse_event(api_db, next_event)
-        return
+        return next_event
 
     if last_ledger_event["message_index"] > last_api_event["message_index"]:
         next_event_sql = (
             "SELECT * FROM messages WHERE message_index > ? ORDER BY message_index ASC LIMIT 1"
         )
         next_event = fetch_one(ledger_db, next_event_sql, (last_api_event["message_index"],))
         parse_event(api_db, next_event)
-        return
+        return next_event
 
     raise exceptions.NoEventToParse("No event to parse")
 
@@ -842,6 +851,12 @@ def synchronize_mempool(api_db, ledger_db):
         pass
 
 
+def refresh_xcp_supply(ledger_db, api_db):
+    xcp_supply = ledger.xcp_supply(ledger_db)
+    cursor = api_db.cursor()
+    cursor.execute("UPDATE assets_info SET supply = ? WHERE asset = 'XCP'", (xcp_supply,))
+
+
 class APIWatcher(Thread):
     def __init__(self):
         Thread.__init__(self)
@@ -895,13 +910,18 @@ def __init__(self):
         self.last_mempool_sync = 0
 
     def follow(self):
+        refresh_xcp_supply(self.ledger_db, self.api_db)
         while not self.stopping and not self.stopped:
+            last_parsed_event = None
             try:
-                parse_next_event(self.api_db, self.ledger_db)
+                last_parsed_event = parse_next_event(self.api_db, self.ledger_db)
             except exceptions.NoEventToParse:
                 logger.trace("API Watcher - No new events to parse")
                 time.sleep(1)
-            if time.time() - self.last_mempool_sync > 10:
+            # let's not sync the mempool when parsing a block
+            if time.time() - self.last_mempool_sync > 10 and (
+                last_parsed_event is None or last_parsed_event["event"] == "BLOCK_PARSED"
+            ):
                 synchronize_mempool(self.api_db, self.ledger_db)
                 self.last_mempool_sync = time.time()
         self.stopped = True

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

@@ -781,6 +781,8 @@ def unpack(db, datahex: str, block_index: int = None):
     :param block_index: Block index of the transaction containing this data
     """
     data = binascii.unhexlify(datahex)
+    if data[: len(config.PREFIX)] == config.PREFIX:
+        data = data[len(config.PREFIX) :]
     message_type_id, message = message_type.unpack(data)
     block_index = block_index or util.CURRENT_BLOCK_INDEX
 

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

@@ -700,18 +700,35 @@ def get_events_by_addresses(
 
 
 def get_all_mempool_events(
-    db, event_name: str = None, cursor: str = None, limit: int = 100, offset: int = None
+    db,
+    event_name: str = None,
+    addresses: str = None,
+    cursor: str = None,
+    limit: int = 100,
+    offset: int = None,
 ):
     """
     Returns all mempool events
     :param str event_name: Comma separated list of events to return
+    :param str addresses: Comma separated list of addresses to return
     :param str cursor: The last event index to return
     :param int limit: The maximum number of events to return (e.g. 5)
     :param int offset: The number of lines to skip before returning results (overrides the `cursor` parameter)
     """
-    where = None
+    where = []
     if event_name:
-        where = [{"event": event} for event in event_name.split(",")]
+        if addresses:
+            for address in addresses.split(","):
+                where = [
+                    {"event": event, "addresses__like": f"%{address}%"}
+                    for event in event_name.split(",")
+                ]
+        else:
+            where = [{"event": event} for event in event_name.split(",")]
+    elif addresses:
+        for address in addresses.split(","):
+            where.append({"addresses__like": f"%{address}%"})
+
     select = "tx_hash, event, bindings AS params, timestamp"
     return select_rows(
         db, "mempool", where=where, last_cursor=cursor, limit=limit, offset=offset, select=select
@@ -1939,7 +1956,7 @@ def get_valid_assets_by_issuer(
     db, address: str, named: bool = None, cursor: str = None, limit: int = 100, offset: int = None
 ):
     """
-    Returns the valid assets of an issuer
+    Returns the valid assets issued by an address
     :param str address: The issuer to return (e.g. $ADDRESS_1)
     :param bool named: Whether to return only named assets (e.g. true)
     :param str cursor: The last index of the assets to return
@@ -1963,6 +1980,64 @@ def get_valid_assets_by_issuer(
     )
 
 
+def get_valid_assets_by_owner(
+    db, address: str, named: bool = None, cursor: str = None, limit: int = 100, offset: int = None
+):
+    """
+    Returns the valid assets owned by an address
+    :param str address: The owner to return (e.g. $ADDRESS_1)
+    :param bool named: Whether to return only named assets (e.g. true)
+    :param str cursor: The last index of the assets to return
+    :param int limit: The maximum number of assets to return (e.g. 5)
+    :param int offset: The number of lines to skip before returning results (overrides the `cursor` parameter)
+    """
+    where = {"owner": address}
+    if named is not None:
+        if named:
+            where["asset__notlike"] = "A%"
+        else:
+            where["asset__like"] = "A%"
+
+    return select_rows(
+        db,
+        "assets_info",
+        where=where,
+        last_cursor=cursor,
+        limit=limit,
+        offset=offset,
+    )
+
+
+def get_valid_assets_by_issuer_or_owner(
+    db, address: str, named: bool = None, cursor: str = None, limit: int = 100, offset: int = None
+):
+    """
+    Returns the valid assets issued or owned by an address
+    :param str address: The issuer or owner to return (e.g. $ADDRESS_1)
+    :param bool named: Whether to return only named assets (e.g. true)
+    :param str cursor: The last index of the assets to return
+    :param int limit: The maximum number of assets to return (e.g. 5)
+    :param int offset: The number of lines to skip before returning results (overrides the `cursor` parameter)
+    """
+    where = [{"issuer": address}, {"owner": address}]
+    if named is not None:
+        if named:
+            for p in where:
+                p["asset__notlike"] = "A%"
+        else:
+            for p in where:
+                p["asset__like"] = "A%"
+
+    return select_rows(
+        db,
+        "assets_info",
+        where=where,
+        last_cursor=cursor,
+        limit=limit,
+        offset=offset,
+    )
+
+
 def get_dividends(db, cursor: str = None, limit: int = 100, offset: int = None):
     """
     Returns all the dividends

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

@@ -65,7 +65,9 @@ def get_routes():
         "/v2/addresses/<address>/dispenses/receives/<asset>": queries.get_dispenses_by_destination_and_asset,
         "/v2/addresses/<address>/sweeps": queries.get_sweeps_by_address,
         "/v2/addresses/<address>/issuances": queries.get_issuances_by_address,
-        "/v2/addresses/<address>/assets": queries.get_valid_assets_by_issuer,
+        "/v2/addresses/<address>/assets": queries.get_valid_assets_by_issuer_or_owner,
+        "/v2/addresses/<address>/assets/issued": queries.get_valid_assets_by_issuer,
+        "/v2/addresses/<address>/assets/owned": queries.get_valid_assets_by_owner,
         "/v2/addresses/<address>/transactions": queries.get_transactions_by_address,
         "/v2/addresses/<address>/dividends": queries.get_dividends_distributed_by_address,
         "/v2/addresses/<address>/orders": queries.get_orders_by_address,

counterparty-core/counterpartycore/lib/blocks.py

@@ -22,6 +22,7 @@
     gas,
     ledger,
     log,
+    mempool,
     message_type,
     util,
 )
@@ -1334,6 +1335,7 @@ def catch_up(db, check_asset_conservation=True):
 
         # Parse the current block
         tx_index = parse_new_block(db, decoded_block, tx_index=tx_index)
+        mempool.clean_mempool(db)
 
         parsed_blocks += 1
         formatted_duration = util.format_duration(time.time() - start_time)

counterparty-core/counterpartycore/lib/check.py

@@ -906,25 +906,26 @@ class SanityError(Exception):
 
 def asset_conservation(db):
     logger.debug("Checking for conservation of assets.")
-    supplies = ledger.supplies(db)
-    held = ledger.held(db)
-    for asset in supplies.keys():
-        asset_issued = supplies[asset]
-        asset_held = held[asset] if asset in held and held[asset] != None else 0  # noqa: E711
-        if asset_issued != asset_held:
-            raise SanityError(
-                "{} {} issued ≠ {} {} held".format(
-                    ledger.value_out(db, asset_issued, asset),
-                    asset,
-                    ledger.value_out(db, asset_held, asset),
-                    asset,
+    with db:
+        supplies = ledger.supplies(db)
+        held = ledger.held(db)
+        for asset in supplies.keys():
+            asset_issued = supplies[asset]
+            asset_held = held[asset] if asset in held and held[asset] != None else 0  # noqa: E711
+            if asset_issued != asset_held:
+                raise SanityError(
+                    "{} {} issued ≠ {} {} held".format(
+                        ledger.value_out(db, asset_issued, asset),
+                        asset,
+                        ledger.value_out(db, asset_held, asset),
+                        asset,
+                    )
+                )
+            logger.trace(
+                "{} has been conserved ({} {} both issued and held)".format(
+                    asset, ledger.value_out(db, asset_issued, asset), asset
                 )
             )
-        logger.debug(
-            "{} has been conserved ({} {} both issued and held)".format(
-                asset, ledger.value_out(db, asset_issued, asset), asset
-            )
-        )
     logger.debug("All assets have been conserved.")
 
 

counterparty-core/counterpartycore/lib/config.py

@@ -5,7 +5,7 @@
 
 
 # Semantic Version
-__version__ = "10.4.2"  # for hatch
+__version__ = "10.4.3"  # for hatch
 VERSION_STRING = __version__
 version = VERSION_STRING.split("-")[0].split(".")
 VERSION_MAJOR = int(version[0])

counterparty-core/counterpartycore/lib/database.py

@@ -116,7 +116,13 @@ def connection(self):
             else:
                 # Too much connections in the pool: closing connection
                 logger.warning("Closing connection due to pool size limit (%s).", self.name)
-                db.close()
+                try:
+                    db.close()
+                except apsw.ThreadingViolationError:
+                    # This should never happen, and yet it has happened..
+                    # let's ignore this harmless error so as not to return a 500 error to the user.
+                    logger.trace("ThreadingViolationError occurred while closing connection.")
+                    pass
 
     def close(self):
         logger.trace(

counterparty-core/counterpartycore/lib/messages/sweep.py

@@ -284,7 +284,13 @@ def parse(db, tx, message):
                             "asset_longname": last_issuance["asset_longname"],
                             "reset": False,
                         }
-                        ledger.insert_record(db, "issuances", bindings, "ASSET_TRANSFER")
+                        ledger.insert_record(
+                            db,
+                            "issuances",
+                            bindings,
+                            "ASSET_TRANSFER",
+                            {"asset_events": "transfer"},
+                        )
                         sweep_pos += 1
 
         bindings = {