Merge pull request #3296 from opentensor/feat/coldkey-swap-clear

Feat/coldkey swap clear

Author: ibraheem-abe

bittensor/core/async_subtensor.py

@@ -56,6 +56,7 @@
 )
 from bittensor.core.extrinsics.asyncex.coldkey_swap import (
     announce_coldkey_swap_extrinsic,
+    clear_coldkey_swap_announcement_extrinsic,
     dispute_coldkey_swap_extrinsic,
     swap_coldkey_announced_extrinsic,
 )
@@ -7115,6 +7116,50 @@ async def dispute_coldkey_swap(
             wait_for_revealed_execution=wait_for_revealed_execution,
         )
 
+    async def clear_coldkey_swap_announcement(
+        self,
+        wallet: "Wallet",
+        *,
+        mev_protection: bool = DEFAULT_MEV_PROTECTION,
+        period: Optional[int] = DEFAULT_PERIOD,
+        raise_error: bool = False,
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = True,
+        wait_for_revealed_execution: bool = True,
+    ) -> ExtrinsicResponse:
+        """
+        Clears (withdraws) a pending coldkey swap announcement.
+
+        Callable by the coldkey that has an active, undisputed swap announcement. The reannouncement delay must have
+        elapsed past the execution block before the announcement can be cleared.
+
+        Parameters:
+            wallet: Bittensor wallet object (should be the current coldkey with an active announcement).
+            mev_protection: If ``True``, encrypts and submits the transaction through the MEV Shield pallet.
+            period: The number of blocks during which the transaction will remain valid.
+            raise_error: Raises a relevant exception rather than returning ``False`` if unsuccessful.
+            wait_for_inclusion: Whether to wait for the inclusion of the transaction.
+            wait_for_finalization: Whether to wait for the finalization of the transaction.
+            wait_for_revealed_execution: Whether to wait for the revealed execution if mev_protection used.
+
+        Returns:
+            ExtrinsicResponse: The result object of the extrinsic execution.
+
+        Notes:
+            - The coldkey must have an active, undisputed swap announcement.
+            - The reannouncement delay must have elapsed past the execution block.
+        """
+        return await clear_coldkey_swap_announcement_extrinsic(
+            subtensor=self,
+            wallet=wallet,
+            mev_protection=mev_protection,
+            period=period,
+            raise_error=raise_error,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+            wait_for_revealed_execution=wait_for_revealed_execution,
+        )
+
     async def dissolve_crowdloan(
         self,
         wallet: "Wallet",

bittensor/core/extrinsics/asyncex/coldkey_swap.py

@@ -184,6 +184,75 @@ async def dispute_coldkey_swap_extrinsic(
         return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)
 
 
+async def clear_coldkey_swap_announcement_extrinsic(
+    subtensor: "AsyncSubtensor",
+    wallet: "Wallet",
+    *,
+    mev_protection: bool = DEFAULT_MEV_PROTECTION,
+    period: Optional[int] = None,
+    raise_error: bool = False,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    wait_for_revealed_execution: bool = True,
+) -> ExtrinsicResponse:
+    """
+    Clears (withdraws) a pending coldkey swap announcement.
+
+    Callable by the coldkey that has an active, undisputed swap announcement. The reannouncement delay must have
+    elapsed past the execution block before the announcement can be cleared.
+
+    Parameters:
+        subtensor: AsyncSubtensor instance with the connection to the chain.
+        wallet: Bittensor wallet object (should be the current coldkey with an active announcement).
+        mev_protection: If ``True``, encrypts and submits the transaction through the MEV Shield pallet.
+        period: The number of blocks during which the transaction will remain valid.
+        raise_error: Raises a relevant exception rather than returning ``False`` if unsuccessful.
+        wait_for_inclusion: Whether to wait for the inclusion of the transaction.
+        wait_for_finalization: Whether to wait for the finalization of the transaction.
+        wait_for_revealed_execution: Whether to wait for the revealed execution if mev_protection used.
+
+    Returns:
+        ExtrinsicResponse: The result object of the extrinsic execution.
+
+    Notes:
+        - The coldkey must have an active, undisputed swap announcement.
+        - The reannouncement delay must have elapsed past the execution block.
+    """
+    try:
+        if not (
+            unlocked := ExtrinsicResponse.unlock_wallet(wallet, raise_error)
+        ).success:
+            return unlocked
+
+        call = await SubtensorModule(subtensor).clear_coldkey_swap_announcement()
+
+        if mev_protection:
+            response = await submit_encrypted_extrinsic(
+                subtensor=subtensor,
+                wallet=wallet,
+                call=call,
+                period=period,
+                raise_error=raise_error,
+                wait_for_inclusion=wait_for_inclusion,
+                wait_for_finalization=wait_for_finalization,
+                wait_for_revealed_execution=wait_for_revealed_execution,
+            )
+        else:
+            response = await subtensor.sign_and_send_extrinsic(
+                call=call,
+                wallet=wallet,
+                wait_for_inclusion=wait_for_inclusion,
+                wait_for_finalization=wait_for_finalization,
+                period=period,
+                raise_error=raise_error,
+            )
+
+        return response
+
+    except Exception as error:
+        return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)
+
+
 async def swap_coldkey_announced_extrinsic(
     subtensor: "AsyncSubtensor",
     wallet: "Wallet",

bittensor/core/extrinsics/coldkey_swap.py

@@ -186,6 +186,75 @@ def dispute_coldkey_swap_extrinsic(
         return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)
 
 
+def clear_coldkey_swap_announcement_extrinsic(
+    subtensor: "Subtensor",
+    wallet: "Wallet",
+    *,
+    mev_protection: bool = DEFAULT_MEV_PROTECTION,
+    period: Optional[int] = None,
+    raise_error: bool = False,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    wait_for_revealed_execution: bool = True,
+) -> ExtrinsicResponse:
+    """
+    Clears (withdraws) a pending coldkey swap announcement.
+
+    Callable by the coldkey that has an active, undisputed swap announcement. The reannouncement delay must have
+    elapsed past the execution block before the announcement can be cleared.
+
+    Parameters:
+        subtensor: Subtensor instance with the connection to the chain.
+        wallet: Bittensor wallet object (should be the current coldkey with an active announcement).
+        mev_protection: If `True`, encrypts and submits the transaction through the MEV Shield pallet.
+        period: The number of blocks during which the transaction will remain valid.
+        raise_error: Raises a relevant exception rather than returning `False` if unsuccessful.
+        wait_for_inclusion: Whether to wait for the inclusion of the transaction.
+        wait_for_finalization: Whether to wait for the finalization of the transaction.
+        wait_for_revealed_execution: Whether to wait for the revealed execution if mev_protection used.
+
+    Returns:
+        ExtrinsicResponse: The result object of the extrinsic execution.
+
+    Notes:
+        - The coldkey must have an active, undisputed swap announcement.
+        - The reannouncement delay must have elapsed past the execution block.
+    """
+    try:
+        if not (
+            unlocked := ExtrinsicResponse.unlock_wallet(wallet, raise_error)
+        ).success:
+            return unlocked
+
+        call = SubtensorModule(subtensor).clear_coldkey_swap_announcement()
+
+        if mev_protection:
+            response = submit_encrypted_extrinsic(
+                subtensor=subtensor,
+                wallet=wallet,
+                call=call,
+                period=period,
+                raise_error=raise_error,
+                wait_for_inclusion=wait_for_inclusion,
+                wait_for_finalization=wait_for_finalization,
+                wait_for_revealed_execution=wait_for_revealed_execution,
+            )
+        else:
+            response = subtensor.sign_and_send_extrinsic(
+                call=call,
+                wallet=wallet,
+                wait_for_inclusion=wait_for_inclusion,
+                wait_for_finalization=wait_for_finalization,
+                period=period,
+                raise_error=raise_error,
+            )
+
+        return response
+
+    except Exception as error:
+        return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)
+
+
 def swap_coldkey_announced_extrinsic(
     subtensor: "Subtensor",
     wallet: "Wallet",

bittensor/core/extrinsics/pallets/subtensor_module.py

@@ -366,6 +366,17 @@ def dispute_coldkey_swap(self) -> Call:
         """
         return self.create_composed_call()
 
+    def clear_coldkey_swap_announcement(self) -> Call:
+        """Returns GenericCall instance for Subtensor function SubtensorModule.clear_coldkey_swap_announcement.
+
+        Callable by the coldkey that has an active swap announcement. Withdraws the announcement
+        after the reannouncement delay has elapsed past the execution block.
+
+        Returns:
+            GenericCall instance.
+        """
+        return self.create_composed_call()
+
     def reset_coldkey_swap(self, coldkey: str) -> Call:
         """Returns GenericCall instance for Subtensor function SubtensorModule.reset_coldkey_swap.
 

bittensor/core/subtensor.py

@@ -55,6 +55,7 @@
 )
 from bittensor.core.extrinsics.coldkey_swap import (
     announce_coldkey_swap_extrinsic,
+    clear_coldkey_swap_announcement_extrinsic,
     dispute_coldkey_swap_extrinsic,
     swap_coldkey_announced_extrinsic,
 )
@@ -5924,6 +5925,50 @@ def dispute_coldkey_swap(
             wait_for_revealed_execution=wait_for_revealed_execution,
         )
 
+    def clear_coldkey_swap_announcement(
+        self,
+        wallet: "Wallet",
+        *,
+        mev_protection: bool = DEFAULT_MEV_PROTECTION,
+        period: Optional[int] = DEFAULT_PERIOD,
+        raise_error: bool = False,
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = True,
+        wait_for_revealed_execution: bool = True,
+    ) -> ExtrinsicResponse:
+        """
+        Clears (withdraws) a pending coldkey swap announcement.
+
+        Callable by the coldkey that has an active, undisputed swap announcement. The reannouncement delay must have
+        elapsed past the execution block before the announcement can be cleared.
+
+        Parameters:
+            wallet: Bittensor wallet object (should be the current coldkey with an active announcement).
+            mev_protection: If `True`, encrypts and submits the transaction through the MEV Shield pallet.
+            period: The number of blocks during which the transaction will remain valid.
+            raise_error: Raises a relevant exception rather than returning `False` if unsuccessful.
+            wait_for_inclusion: Whether to wait for the inclusion of the transaction.
+            wait_for_finalization: Whether to wait for the finalization of the transaction.
+            wait_for_revealed_execution: Whether to wait for the revealed execution if mev_protection used.
+
+        Returns:
+            ExtrinsicResponse: The result object of the extrinsic execution.
+
+        Notes:
+            - The coldkey must have an active, undisputed swap announcement.
+            - The reannouncement delay must have elapsed past the execution block.
+        """
+        return clear_coldkey_swap_announcement_extrinsic(
+            subtensor=self,
+            wallet=wallet,
+            mev_protection=mev_protection,
+            period=period,
+            raise_error=raise_error,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+            wait_for_revealed_execution=wait_for_revealed_execution,
+        )
+
     def dissolve_crowdloan(
         self,
         wallet: "Wallet",

bittensor/extras/subtensor_api/extrinsics.py

@@ -13,6 +13,7 @@ def __init__(self, subtensor: Union["_Subtensor", "_AsyncSubtensor"]):
         self.add_stake_burn = subtensor.add_stake_burn
         self.add_stake_multiple = subtensor.add_stake_multiple
         self.announce_coldkey_swap = subtensor.announce_coldkey_swap
+        self.clear_coldkey_swap_announcement = subtensor.clear_coldkey_swap_announcement
         self.dispute_coldkey_swap = subtensor.dispute_coldkey_swap
         self.burned_register = subtensor.burned_register
         self.claim_root = subtensor.claim_root