refactor(nano): implement BlueprintService

Author: glevco

hathor/builder/builder.py

@@ -35,7 +35,7 @@
 from hathor.manager import HathorManager
 from hathor.mining.cpu_mining_service import CpuMiningService
 from hathor.nanocontracts import NCRocksDBStorageFactory, NCStorageFactory
-from hathor.nanocontracts.catalog import NCBlueprintCatalog
+from hathor.nanocontracts.blueprint_service import BlueprintService
 from hathor.nanocontracts.nc_exec_logs import NCLogConfig, NCLogStorage
 from hathor.nanocontracts.runner.runner import RunnerFactory
 from hathor.nanocontracts.sorter.types import NCSorterCallable
@@ -110,7 +110,7 @@ class BuildArtifacts(NamedTuple):
 
 
 _VertexVerifiersBuilder: TypeAlias = Callable[
-    [Reactor, HathorSettingsType, DifficultyAdjustmentAlgorithm, FeatureService, TransactionStorage],
+    [Reactor, HathorSettingsType, DifficultyAdjustmentAlgorithm, FeatureService, TransactionStorage, BlueprintService],
     VertexVerifiers
 ]
 
@@ -199,6 +199,7 @@ def __init__(self) -> None:
         self._nc_log_config: NCLogConfig = NCLogConfig.NONE
 
         self._vertex_json_serializer: VertexJsonSerializer | None = None
+        self._blueprint_service: BlueprintService | None = None
 
     def build(self) -> BuildArtifacts:
         if self.artifacts is not None:
@@ -233,9 +234,7 @@ def build(self) -> BuildArtifacts:
         poa_block_producer = self._get_or_create_poa_block_producer()
         runner_factory = self._get_or_create_runner_factory()
         vertex_json_serializer = self._get_or_create_vertex_json_serializer()
-
-        if settings.ENABLE_NANO_CONTRACTS:
-            tx_storage.nc_catalog = self._get_nc_catalog()
+        blueprint_service = self._get_or_create_blueprint_service()
 
         if self._enable_address_index:
             indexes.enable_address_index(pubsub)
@@ -279,6 +278,7 @@ def build(self) -> BuildArtifacts:
             runner_factory=runner_factory,
             feature_service=feature_service,
             vertex_json_serializer=vertex_json_serializer,
+            blueprint_service=blueprint_service,
             **kwargs
         )
 
@@ -426,18 +426,14 @@ def _get_or_create_consensus(self) -> ConsensusAlgorithm:
 
         return self._consensus
 
-    def _get_nc_catalog(self) -> NCBlueprintCatalog:
-        from hathor.nanocontracts.catalog import generate_catalog_from_settings
-        settings = self._get_or_create_settings()
-        return generate_catalog_from_settings(settings)
-
     def _get_or_create_runner_factory(self) -> RunnerFactory:
         if self._runner_factory is None:
             self._runner_factory = RunnerFactory(
                 reactor=self._get_reactor(),
                 settings=self._get_or_create_settings(),
                 tx_storage=self._get_or_create_tx_storage(),
                 nc_storage_factory=self._get_or_create_nc_storage_factory(),
+                blueprint_service=self._get_or_create_blueprint_service(),
             )
         return self._runner_factory
 
@@ -611,14 +607,16 @@ def _get_or_create_vertex_verifiers(self) -> VertexVerifiers:
             feature_service = self._get_or_create_feature_service()
             daa = self._get_or_create_daa()
             tx_storage = self._get_or_create_tx_storage()
+            blueprint_service = self._get_or_create_blueprint_service()
 
             if self._vertex_verifiers_builder:
                 self._vertex_verifiers = self._vertex_verifiers_builder(
                     reactor,
                     settings,
                     daa,
                     feature_service,
-                    tx_storage
+                    tx_storage,
+                    blueprint_service,
                 )
             else:
                 self._vertex_verifiers = VertexVerifiers.create_defaults(
@@ -627,6 +625,7 @@ def _get_or_create_vertex_verifiers(self) -> VertexVerifiers:
                     daa=daa,
                     feature_service=feature_service,
                     tx_storage=tx_storage,
+                    blueprint_service=blueprint_service,
                 )
 
         return self._vertex_verifiers
@@ -684,13 +683,25 @@ def _get_or_create_vertex_json_serializer(self) -> VertexJsonSerializer:
         if self._vertex_json_serializer is None:
             tx_storage = self._get_or_create_tx_storage()
             nc_log_storage = self._get_or_create_nc_log_storage()
+            blueprint_service = self._get_or_create_blueprint_service()
             self._vertex_json_serializer = VertexJsonSerializer(
                 storage=tx_storage,
                 nc_log_storage=nc_log_storage,
+                blueprint_service=blueprint_service,
             )
 
         return self._vertex_json_serializer
 
+    def _get_or_create_blueprint_service(self) -> BlueprintService:
+        if self._blueprint_service is None:
+            self._blueprint_service = BlueprintService(
+                settings=self._get_or_create_settings(),
+                tx_storage=self._get_or_create_tx_storage(),
+                feature_service=self._get_or_create_feature_service(),
+            )
+
+        return self._blueprint_service
+
     def set_rocksdb_path(self, path: str | tempfile.TemporaryDirectory) -> 'Builder':
         if self._tx_storage:
             raise ValueError('cannot set rocksdb path after tx storage is set')

hathor/dag_builder/builder.py

@@ -88,14 +88,13 @@ def from_manager(
         blueprints_module: ModuleType | None = None
     ) -> DAGBuilder:
         """Create a DAGBuilder instance from a HathorManager instance."""
-        assert manager.tx_storage.nc_catalog
         return DAGBuilder(
             settings=manager._settings,
             daa=manager.daa,
             genesis_wallet=initialize_hd_wallet(genesis_words),
             wallet_factory=wallet_factory,
             vertex_resolver=lambda x: manager.cpu_mining_service.resolve(x),
-            nc_catalog=manager.tx_storage.nc_catalog,
+            nc_catalog=manager.blueprint_service.nc_catalog,
             blueprints_module=blueprints_module,
         )
 

hathor/dag_builder/cli.py

@@ -23,7 +23,7 @@ def main(filename: str, genesis_seed: str) -> None:
 
     from hathor.conf.get_settings import get_global_settings
     from hathor.daa import DifficultyAdjustmentAlgorithm
-    from hathor.nanocontracts.catalog import generate_catalog_from_settings
+    from hathor.nanocontracts.catalog import NCBlueprintCatalog
     from hathor.wallet import HDWallet
     settings = get_global_settings()
 
@@ -37,7 +37,9 @@ def wallet_factory(words=None):
 
     genesis_wallet = wallet_factory(genesis_seed)
     daa = DifficultyAdjustmentAlgorithm(settings=settings)
-    nc_catalog = generate_catalog_from_settings(settings)
+    nc_catalog = NCBlueprintCatalog()
+    blueprints = NCBlueprintCatalog.generate_blueprints_from_settings(settings)
+    nc_catalog.register_blueprints(blueprints)
 
     builder = DAGBuilder(
         settings=settings,

hathor/manager.py

@@ -46,6 +46,7 @@
 from hathor.feature_activation.utils import Features
 from hathor.mining import BlockTemplate, BlockTemplates
 from hathor.mining.cpu_mining_service import CpuMiningService
+from hathor.nanocontracts.blueprint_service import BlueprintService
 from hathor.nanocontracts.runner import Runner
 from hathor.nanocontracts.runner.runner import RunnerFactory
 from hathor.nanocontracts.storage import NCBlockStorage, NCContractStorage, get_block_storage_from_block
@@ -117,6 +118,7 @@ def __init__(
         runner_factory: RunnerFactory,
         feature_service: FeatureService,
         vertex_json_serializer: VertexJsonSerializer,
+        blueprint_service: BlueprintService,
         hostname: Optional[str] = None,
         wallet: Optional[BaseWallet] = None,
         capabilities: Optional[list[str]] = None,
@@ -207,6 +209,7 @@ def __init__(
         self.runner_factory = runner_factory
         self.feature_service = feature_service
         self.vertex_json_serializer = vertex_json_serializer
+        self.blueprint_service = blueprint_service
 
         self.websocket_factory = websocket_factory
 

hathor/nanocontracts/blueprint_service.py

@@ -0,0 +1,118 @@
+#  Copyright 2026 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from structlog import get_logger
+
+from hathor.conf.settings import HathorSettings
+from hathor.feature_activation.feature_service import FeatureService
+from hathor.nanocontracts import OnChainBlueprint
+from hathor.nanocontracts.catalog import NCBlueprintCatalog
+from hathor.nanocontracts.exception import (
+    BlueprintDoesNotExist,
+    OCBBlueprintNotConfirmed,
+    OCBInvalidBlueprintVertexType,
+)
+from hathor.transaction.storage import TransactionStorage
+from hathor.transaction.storage.exceptions import TransactionDoesNotExist
+from hathorlib.nanocontracts.types import BlueprintId
+
+if TYPE_CHECKING:
+    from hathor import Blueprint
+
+logger = get_logger()
+
+
+class BlueprintService:
+    __slots__ = ('log', 'settings', 'nc_catalog', 'tx_storage', 'feature_service')
+
+    def __init__(
+        self,
+        *,
+        settings: HathorSettings,
+        tx_storage: TransactionStorage,
+        feature_service: FeatureService,
+    ) -> None:
+        self.log = logger.new()
+        self.settings = settings
+        self.nc_catalog = NCBlueprintCatalog()
+        self.tx_storage = tx_storage
+        self.feature_service = feature_service
+
+        if settings.ENABLE_NANO_CONTRACTS:
+            blueprints = NCBlueprintCatalog.generate_blueprints_from_settings(settings)
+            self.register_blueprints(blueprints)
+
+    def get_on_chain_blueprint(self, blueprint_id: BlueprintId) -> OnChainBlueprint:
+        """Return an on-chain blueprint transaction."""
+        try:
+            blueprint_tx = self.tx_storage.get_transaction(blueprint_id)
+        except TransactionDoesNotExist:
+            self.log.debug('no transaction with the given id found', blueprint_id=blueprint_id.hex())
+            raise BlueprintDoesNotExist(blueprint_id.hex())
+        if not isinstance(blueprint_tx, OnChainBlueprint):
+            raise OCBInvalidBlueprintVertexType(blueprint_id.hex())
+        tx_meta = blueprint_tx.get_metadata()
+        if tx_meta.voided_by or not tx_meta.first_block:
+            raise OCBBlueprintNotConfirmed(blueprint_id.hex())
+        # XXX: maybe use N blocks confirmation, like reward-locks
+        return blueprint_tx
+
+    def get_blueprint_class(self, blueprint_id: BlueprintId) -> type[Blueprint]:
+        """Returns the blueprint class associated with the given blueprint_id.
+
+        The blueprint class could be in the catalog (first search), or it could be the tx_id of an on-chain blueprint.
+        """
+        from hathor.nanocontracts import OnChainBlueprint
+        blueprint = self._get_blueprint(blueprint_id)
+        if isinstance(blueprint, OnChainBlueprint):
+            return blueprint.get_blueprint_class()
+        else:
+            return blueprint
+
+    def get_blueprint_source(self, blueprint_id: BlueprintId) -> str:
+        """Returns the source code associated with the given blueprint_id.
+
+        The blueprint class could be in the catalog (first search), or it could be the tx_id of an on-chain blueprint.
+        """
+        import inspect
+
+        from hathor.nanocontracts import OnChainBlueprint
+
+        blueprint = self._get_blueprint(blueprint_id)
+        if isinstance(blueprint, OnChainBlueprint):
+            return self.get_on_chain_blueprint(blueprint_id).code.text
+        else:
+            module = inspect.getmodule(blueprint)
+            assert module is not None
+            return inspect.getsource(module)
+
+    def _get_blueprint(self, blueprint_id: BlueprintId) -> type[Blueprint] | OnChainBlueprint:
+        if blueprint_class := self.nc_catalog.get_blueprint_class(blueprint_id):
+            return blueprint_class
+
+        self.log.debug(
+            'blueprint_id not in the catalog, looking for on-chain blueprint',
+            blueprint_id=blueprint_id.hex()
+        )
+        return self.get_on_chain_blueprint(blueprint_id)
+
+    def register_blueprint(self, blueprint_id: bytes, blueprint: type[Blueprint], *, strict: bool = False) -> None:
+        self.nc_catalog.register_blueprints({blueprint_id: blueprint}, strict=strict)
+
+    def register_blueprints(self, blueprints: dict[bytes, type[Blueprint]], *, strict: bool = False) -> None:
+        self.nc_catalog.register_blueprints(blueprints, strict=strict)

hathor/nanocontracts/blueprints/__init__.py

@@ -1,21 +0,0 @@
-# Copyright 2023 Hathor Labs
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from typing import TYPE_CHECKING, Type
-
-if TYPE_CHECKING:
-    from hathor.nanocontracts.blueprint import Blueprint
-
-
-_blueprints_mapper: dict[str, Type['Blueprint']] = {}

hathor/nanocontracts/catalog.py

@@ -12,31 +12,46 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import TYPE_CHECKING, Type
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
-from hathor.nanocontracts.blueprints import _blueprints_mapper
 from hathor.nanocontracts.types import BlueprintId
 
 if TYPE_CHECKING:
     from hathor.conf.settings import HathorSettings
     from hathor.nanocontracts.blueprint import Blueprint
 
 
+_BLUEPRINTS_MAPPER: dict[str, type[Blueprint]] = {}
+
+
 class NCBlueprintCatalog:
     """Catalog of blueprints available."""
+    __slots__ = ('_blueprints',)
 
-    def __init__(self, blueprints: dict[bytes, Type['Blueprint']]) -> None:
-        self.blueprints = blueprints
+    def __init__(self) -> None:
+        self._blueprints: dict[bytes, type[Blueprint]] = {}
 
-    def get_blueprint_class(self, blueprint_id: BlueprintId) -> Type['Blueprint'] | None:
+    def get_blueprint_class(self, blueprint_id: BlueprintId) -> type[Blueprint] | None:
         """Return the blueprint class related to the given blueprint id or None if it doesn't exist."""
-        return self.blueprints.get(blueprint_id, None)
-
-
-def generate_catalog_from_settings(settings: 'HathorSettings') -> NCBlueprintCatalog:
-    """Generate a catalog of blueprints based on the provided settings."""
-    assert settings.ENABLE_NANO_CONTRACTS
-    blueprints: dict[bytes, Type['Blueprint']] = {}
-    for _id, _name in settings.BLUEPRINTS.items():
-        blueprints[_id] = _blueprints_mapper[_name]
-    return NCBlueprintCatalog(blueprints)
+        return self._blueprints.get(blueprint_id)
+
+    def register_blueprints(self, blueprints: dict[bytes, type[Blueprint]], *, strict: bool = False) -> None:
+        if strict:
+            for blueprint_id in blueprints:
+                if blueprint := self._blueprints.get(blueprint_id):
+                    raise ValueError(f'Blueprint {blueprint_id.hex()} is already registered: {blueprint.__name__}')
+        self._blueprints.update(blueprints)
+
+    def get_all(self) -> dict[bytes, type[Blueprint]]:
+        return dict(self._blueprints)
+
+    @staticmethod
+    def generate_blueprints_from_settings(settings: HathorSettings) -> dict[bytes, type[Blueprint]]:
+        """Generate a map of blueprints based on the provided settings."""
+        assert settings.ENABLE_NANO_CONTRACTS
+        blueprints: dict[bytes, type[Blueprint]] = {}
+        for id_, name in settings.BLUEPRINTS.items():
+            blueprints[id_] = _BLUEPRINTS_MAPPER[name]
+        return blueprints

hathor/nanocontracts/resources/blueprint.py

@@ -92,7 +92,7 @@ def render_GET(self, request: 'Request') -> bytes:
             return error_response.json_dumpb()
 
         try:
-            blueprint_class = self.manager.tx_storage.get_blueprint_class(blueprint_id)
+            blueprint_class = self.manager.blueprint_service.get_blueprint_class(blueprint_id)
         except BlueprintDoesNotExist:
             request.setResponseCode(404)
             error_response = ErrorResponse(success=False, error=f'Blueprint not found: {params.blueprint_id}')