stub chiapos2 API

Author: arvidn

chia/_tests/core/custom_types/test_proof_of_space.py

@@ -12,7 +12,7 @@
 from chia._tests.util.misc import Marks, datacases
 from chia.consensus.default_constants import DEFAULT_CONSTANTS
 from chia.types.blockchain_format.proof_of_space import (
-    calculate_plot_difficulty,
+    calculate_plot_strength,
     calculate_prefix_bits,
     check_plot_size,
     make_pos,
@@ -201,7 +201,7 @@ def test_verify_and_get_quality_string_v2(caplog: pytest.LogCaptureFixture, case
 
 
 @pytest.mark.parametrize(
-    "height, difficulty",
+    "height, strength",
     [
         (0, 2),
         (DEFAULT_CONSTANTS.HARD_FORK_HEIGHT, 2),
@@ -219,8 +219,8 @@ def test_verify_and_get_quality_string_v2(caplog: pytest.LogCaptureFixture, case
         (DEFAULT_CONSTANTS.PLOT_DIFFICULTY_8_HEIGHT + 1000000, 8),
     ],
 )
-def test_calculate_plot_difficulty(height: uint32, difficulty: uint8) -> None:
-    assert calculate_plot_difficulty(DEFAULT_CONSTANTS, height) == difficulty
+def test_calculate_plot_strength(height: uint32, strength: uint8) -> None:
+    assert calculate_plot_strength(DEFAULT_CONSTANTS, height) == strength
 
 
 @pytest.mark.parametrize(

chia/_tests/plotting/test_prover.py

@@ -5,6 +5,8 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from chia_rs.sized_bytes import bytes32
+from chia_rs.sized_ints import uint8
 
 from chia.plotting.prover import PlotVersion, V1Prover, V2Prover, get_prover_from_bytes, get_prover_from_file
 
@@ -27,8 +29,7 @@ def test_v2_prover_get_memo_raises_error(self) -> None:
 
     def test_v2_prover_get_compression_level_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")
-        with pytest.raises(NotImplementedError, match="V2 plot format is not yet implemented"):
-            prover.get_compression_level()
+        assert prover.get_compression_level() == uint8(0)
 
     def test_v2_prover_get_id_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")
@@ -38,12 +39,12 @@ def test_v2_prover_get_id_raises_error(self) -> None:
     def test_v2_prover_get_qualities_for_challenge_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")
         with pytest.raises(NotImplementedError, match="V2 plot format is not yet implemented"):
-            prover.get_qualities_for_challenge(b"challenge")
+            prover.get_qualities_for_challenge(bytes32.random(), uint8(2))
 
     def test_v2_prover_get_full_proof_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")
-        with pytest.raises(NotImplementedError, match="V2 plot format is not yet implemented"):
-            prover.get_full_proof(b"challenge", 0)
+        with pytest.raises(AssertionError, match="V2 plots don't have full proofs"):
+            prover.get_full_proof(bytes32.random(), 0)
 
     def test_v2_prover_bytes_raises_error(self) -> None:
         prover = V2Prover("/nonexistent/path/test.plot2")

chia/harvester/harvester_api.py

@@ -24,6 +24,7 @@
 from chia.server.api_protocol import ApiMetadata
 from chia.server.ws_connection import WSChiaConnection
 from chia.types.blockchain_format.proof_of_space import (
+    calculate_plot_strength,
     calculate_pos_challenge,
     calculate_prefix_bits,
     generate_plot_public_key,
@@ -95,6 +96,8 @@ async def new_signage_point_harvester(
         start = time.monotonic()
         assert len(new_challenge.challenge_hash) == 32
 
+        plot_strength = calculate_plot_strength(self.harvester.constants, new_challenge.last_tx_height)
+
         loop = asyncio.get_running_loop()
 
         def blocking_lookup(filename: Path, plot_info: PlotInfo) -> list[tuple[bytes32, ProofOfSpace]]:
@@ -108,7 +111,7 @@ def blocking_lookup(filename: Path, plot_info: PlotInfo) -> list[tuple[bytes32,
                     new_challenge.sp_hash,
                 )
                 try:
-                    quality_strings = plot_info.prover.get_qualities_for_challenge(sp_challenge_hash)
+                    quality_strings = plot_info.prover.get_qualities_for_challenge(sp_challenge_hash, plot_strength)
                 except RuntimeError as e:
                     if str(e) == "Timeout waiting for context queue.":
                         self.harvester.log.warning(

chia/plotting/check_plots.py

@@ -12,6 +12,7 @@
 from chia_rs.sized_ints import uint8, uint32
 from chiapos import Verifier
 
+from chia.consensus.default_constants import DEFAULT_CONSTANTS
 from chia.plotting.manager import PlotManager
 from chia.plotting.util import (
     PlotInfo,
@@ -173,7 +174,9 @@ def process_plot(plot_path: Path, plot_info: PlotInfo, num_start: int, num_end:
                 # Some plot errors cause get_qualities_for_challenge to throw a RuntimeError
                 try:
                     quality_start_time = round(time() * 1000)
-                    for index, quality_str in enumerate(pr.get_qualities_for_challenge(challenge)):
+                    for index, quality_str in enumerate(
+                        pr.get_qualities_for_challenge(challenge, DEFAULT_CONSTANTS.PLOT_DIFFICULTY_INITIAL)
+                    ):
                         quality_spent_time = round(time() * 1000) - quality_start_time
                         if quality_spent_time > 8000:
                             log.warning(

chia/plotting/prover.py

@@ -8,6 +8,8 @@
 from chia_rs.sized_ints import uint8
 from chiapos import DiskProver
 
+from chia.types.blockchain_format.proof_of_space import quality_for_partial_proof
+
 if TYPE_CHECKING:
     from chiapos import DiskProver
 
@@ -27,8 +29,14 @@ def get_compression_level(self) -> uint8: ...
     def get_version(self) -> PlotVersion: ...
     def __bytes__(self) -> bytes: ...
     def get_id(self) -> bytes32: ...
-    def get_qualities_for_challenge(self, challenge: bytes32) -> list[bytes32]: ...
-    def get_full_proof(self, challenge: bytes, index: int, parallel_read: bool = True) -> bytes: ...
+    def get_qualities_for_challenge(self, challenge: bytes32, plot_strength: uint8) -> list[bytes32]: ...
+
+    # this is only supported by v2 plots
+    def get_partial_proofs_for_challenge(self, challenge: bytes32, plot_strength: uint8) -> list[bytes]: ...
+
+    # this is only supported by v1 plots. v2 plots first get the partial proof
+    # and turn it into a full proof by calling solve_proof(), or pass it to the solver service
+    def get_full_proof(self, challenge: bytes32, index: int, parallel_read: bool = True) -> bytes: ...
 
     @classmethod
     def from_bytes(cls, data: bytes) -> ProverProtocol: ...
@@ -55,8 +63,8 @@ def get_memo(self) -> bytes:
         raise NotImplementedError("V2 plot format is not yet implemented")
 
     def get_compression_level(self) -> uint8:
-        # TODO: todo_v2_plots implement compression level retrieval
-        raise NotImplementedError("V2 plot format is not yet implemented")
+        # v2 plots do not support compression
+        return uint8(0)
 
     def get_version(self) -> PlotVersion:
         return PlotVersion.V2
@@ -69,13 +77,18 @@ def get_id(self) -> bytes32:
         # TODO: Extract plot ID from V2 plot file
         raise NotImplementedError("V2 plot format is not yet implemented")
 
-    def get_qualities_for_challenge(self, challenge: bytes) -> list[bytes32]:
+    def get_partial_proofs_for_challenge(self, challenge: bytes, plot_strength: uint8) -> list[bytes]:
         # TODO: todo_v2_plots Implement plot quality lookup
         raise NotImplementedError("V2 plot format is not yet implemented")
 
-    def get_full_proof(self, challenge: bytes, index: int, parallel_read: bool = True) -> bytes:
-        # TODO: todo_v2_plots Implement plot proof generation
-        raise NotImplementedError("V2 plot format is not yet implemented")
+    def get_qualities_for_challenge(self, challenge: bytes32, plot_strength: uint8) -> list[bytes32]:
+        return [
+            quality_for_partial_proof(partial, challenge)
+            for partial in self.get_partial_proofs_for_challenge(challenge, plot_strength)
+        ]
+
+    def get_full_proof(self, challenge: bytes32, index: int, parallel_read: bool = True) -> bytes:
+        raise AssertionError("V2 plots don't have full proofs")
 
     @classmethod
     def from_bytes(cls, data: bytes) -> V2Prover:
@@ -113,10 +126,13 @@ def __bytes__(self) -> bytes:
     def get_id(self) -> bytes32:
         return bytes32(self._disk_prover.get_id())
 
-    def get_qualities_for_challenge(self, challenge: bytes32) -> list[bytes32]:
+    def get_partial_proofs_for_challenge(self, challenge: bytes32, plot_strength: uint8) -> list[bytes]:
+        raise AssertionError("V1 plot format doesn't use partial proofs")
+
+    def get_qualities_for_challenge(self, challenge: bytes32, plot_strength: uint8) -> list[bytes32]:
         return [bytes32(quality) for quality in self._disk_prover.get_qualities_for_challenge(challenge)]
 
-    def get_full_proof(self, challenge: bytes, index: int, parallel_read: bool = True) -> bytes:
+    def get_full_proof(self, challenge: bytes32, index: int, parallel_read: bool = True) -> bytes:
         return bytes(self._disk_prover.get_full_proof(challenge, index, parallel_read))
 
     @classmethod

chia/simulator/block_tools.py

@@ -89,6 +89,7 @@
 from chia.types.blockchain_format.coin import Coin
 from chia.types.blockchain_format.program import DEFAULT_FLAGS, INFINITE_COST, Program, _run, run_with_cost
 from chia.types.blockchain_format.proof_of_space import (
+    calculate_plot_strength,
     calculate_pos_challenge,
     calculate_prefix_bits,
     generate_plot_public_key,
@@ -1500,6 +1501,8 @@ def get_pospaces_for_challenge(
         found_proofs: list[tuple[uint64, ProofOfSpace]] = []
         rng = random.Random()
         rng.seed(seed)
+
+        plot_strength = calculate_plot_strength(constants, height)
         for plot_info in self.plot_manager.plots.values():
             plot_id: bytes32 = plot_info.prover.get_id()
             if force_plot_id is not None and plot_id != force_plot_id:
@@ -1509,7 +1512,7 @@ def get_pospaces_for_challenge(
                 continue
 
             new_challenge: bytes32 = calculate_pos_challenge(plot_id, challenge_hash, signage_point)
-            qualities = plot_info.prover.get_qualities_for_challenge(new_challenge)
+            qualities = plot_info.prover.get_qualities_for_challenge(new_challenge, plot_strength)
 
             for proof_index, quality_str in enumerate(qualities):
                 required_iters = calculate_iterations_quality(

chia/types/blockchain_format/proof_of_space.py

@@ -13,6 +13,28 @@
 
 log = logging.getLogger(__name__)
 
+# These are temporary stubs for chiapos2, that we build against until it's ready to be integrated.
+
+
+# returns quality string for v2 plot, or None if invalid
+def validate_proof_v2(
+    plot_id: bytes32, size: uint8, plot_strength: uint8, challenge: bytes32, proof: bytes
+) -> Optional[bytes32]:
+    # TODO: todo_v2_plots call into new chiapos library
+    raise NotImplementedError
+
+
+# this is compute intensive, solving a partial proof returning a full proof
+def solve_proof(partial_proof: bytes) -> bytes:
+    # TODO: todo_v2_plots call into new chiapos library
+    raise NotImplementedError
+
+
+# given a partial proof, computes the quality. This is used to compute required iters.
+def quality_for_partial_proof(partial_proof: bytes, challenge: bytes32) -> bytes32:
+    # TODO: todo_v2_plots call into new chiapos library
+    return std_hash(partial_proof + challenge)
+
 
 def make_pos(
     challenge: bytes32,
@@ -49,14 +71,6 @@ def get_plot_id(pos: ProofOfSpace) -> bytes32:
     return calculate_plot_id_pk(pos.pool_public_key, pos.plot_public_key)
 
 
-# returns quality string for v2 plot, or None if invalid
-def validate_proof_v2(
-    plot_id: bytes32, size: uint8, difficulty: uint8, challenge: bytes32, proof: bytes
-) -> Optional[bytes32]:
-    # TODO: todo_v2_plots call into new chiapos library
-    raise NotImplementedError
-
-
 def check_plot_size(constants: ConsensusConstants, ps: PlotSize) -> bool:
     size_v1 = ps.size_v1
     if size_v1 is not None:
@@ -128,8 +142,8 @@ def verify_and_get_quality_string(
         # === V2 plots ===
         assert plot_size.size_v2 is not None
 
-        plot_difficulty = calculate_plot_difficulty(constants, height)
-        return validate_proof_v2(plot_id, plot_size.size_v2, plot_difficulty, pos.challenge, bytes(pos.proof))
+        plot_strength = calculate_plot_strength(constants, height)
+        return validate_proof_v2(plot_id, plot_size.size_v2, plot_strength, pos.challenge, bytes(pos.proof))
 
 
 def passes_plot_filter(
@@ -167,7 +181,7 @@ def calculate_prefix_bits(constants: ConsensusConstants, height: uint32, plot_si
     return max(0, prefix_bits)
 
 
-def calculate_plot_difficulty(constants: ConsensusConstants, height: uint32) -> uint8:
+def calculate_plot_strength(constants: ConsensusConstants, height: uint32) -> uint8:
     if height < constants.PLOT_DIFFICULTY_4_HEIGHT:
         return constants.PLOT_DIFFICULTY_INITIAL
     if height < constants.PLOT_DIFFICULTY_5_HEIGHT: