Skip to content

taker.podle_manager

taker.podle_manager

Manager for PoDLE commitments (used for retry tracking).

Classes

PoDLEManager

Manages tracking of used PoDLE commitments.

Source code in taker/src/taker/podle_manager.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class PoDLEManager:
    """Manages tracking of used PoDLE commitments."""

    def __init__(self, data_dir: Path | None = None):
        self.filepath = get_used_commitments_path(data_dir)
        self.used_commitments: set[str] = set()
        self.external_commitments: dict = {}
        self._load()

    def _load(self) -> None:
        """Load used commitments from file."""
        if not self.filepath.exists():
            return
        try:
            with open(self.filepath) as f:
                data = json.load(f)
                # Handle reference implementation format: {"used": ["hex..."], "external": ...}
                if isinstance(data, dict):
                    self.used_commitments = set(data.get("used", []))
                    self.external_commitments = data.get("external", {})
                else:
                    self.used_commitments = set()
                    self.external_commitments = {}
            logger.debug(f"Loaded {len(self.used_commitments)} used PoDLE commitments")
        except Exception as e:
            logger.error(f"Failed to load used commitments: {e}")

    def _save(self) -> None:
        """Save used commitments to file."""
        try:
            data = {
                "used": list(self.used_commitments),
                "external": self.external_commitments,
            }
            with open(self.filepath, "w") as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            logger.error(f"Failed to save used commitments: {e}")

    def get_utxo_retry_count(self, utxo_str: str, private_key: bytes, max_retries: int) -> int:
        """
        Get the number of times a UTXO has been used for PoDLE commitments.

        Checks indices 0..(max_retries-1) in reverse order and returns the highest
        index + 1 where a commitment is found in used_commitments.

        Note: Only used in tests. Production code uses lazy evaluation in
        generate_fresh_commitment() to avoid generating all commitments upfront.

        Returns:
            0 if UTXO is fresh (no used commitments)
            1-max_retries if UTXO has been used that many times
        """
        # Early termination: stop at first match (reverse order)
        for i in reversed(range(max_retries)):
            try:
                podle = generate_podle(private_key, utxo_str, i)
                commitment_hex = podle.commitment.hex()
                if commitment_hex in self.used_commitments:
                    return i + 1  # Found highest used index
            except Exception:
                continue
        return 0  # No used commitments found

    def generate_fresh_commitment(
        self,
        wallet_utxos: list[UTXOInfo],
        cj_amount: int,
        private_key_getter: Any,  # Callable[[str], bytes]
        min_confirmations: int = 5,
        min_percent: int = 20,
        max_retries: int = 3,
    ) -> ExtendedPoDLECommitment | None:
        """
        Generate a fresh PoDLE commitment for a CoinJoin.

        Iterates through eligible UTXOs and tries indices 0..max_retries-1 until
        finding an unused commitment. UTXOs are pre-sorted by confirmations and value,
        so fresh UTXOs (which succeed at index 0) are naturally preferred.

        Args:
            wallet_utxos: Available wallet UTXOs
            cj_amount: CoinJoin amount
            private_key_getter: Function to get private key for address
            min_confirmations: Minimum UTXO confirmations required
            min_percent: Minimum UTXO value as % of cj_amount
            max_retries: Maximum number of retries per UTXO (default: 3)

        Returns:
            ExtendedPoDLECommitment or None if no fresh commitment available
        """
        eligible_utxos = get_eligible_podle_utxos(
            wallet_utxos, cj_amount, min_confirmations, min_percent
        )

        if not eligible_utxos:
            logger.warning("No eligible UTXOs for PoDLE")
            return None

        # Try each UTXO in order (already sorted by confirmations, value)
        # Fresh UTXOs naturally succeed faster (at index 0)
        for utxo in eligible_utxos:
            private_key = private_key_getter(utxo.address)
            if private_key is None:
                continue

            utxo_str = f"{utxo.txid}:{utxo.vout}"

            # Try indices 0..max_retries-1 for this UTXO
            for index in range(max_retries):
                try:
                    # Generate commitment to check hash
                    podle = generate_podle(private_key, utxo_str, index)
                    commitment_hex = podle.commitment.hex()

                    if commitment_hex in self.used_commitments:
                        logger.debug(f"PoDLE commitment for {utxo_str} index {index} already used")
                        continue

                    # Found unused commitment
                    self.used_commitments.add(commitment_hex)
                    self._save()

                    logger.info(
                        f"Generated fresh PoDLE for {utxo_str} using index {index} "
                        f"(utxo value={utxo.value}, confs={utxo.confirmations})"
                    )

                    return ExtendedPoDLECommitment(
                        commitment=podle,
                        scriptpubkey=utxo.scriptpubkey,
                        blockheight=utxo.height,
                    )
                except Exception as e:
                    logger.warning(f"Failed to generate PoDLE for {utxo_str} index {index}: {e}")
                    continue

            # All indices exhausted for this UTXO
            logger.debug(f"Skipping {utxo.txid}:{utxo.vout} - all {max_retries} indices used")

        logger.error("Failed to generate any fresh PoDLE commitment from available UTXOs")
        return None
Attributes
external_commitments: dict = {} instance-attribute
filepath = get_used_commitments_path(data_dir) instance-attribute
used_commitments: set[str] = set() instance-attribute
Functions
__init__(data_dir: Path | None = None)
Source code in taker/src/taker/podle_manager.py 
24
25
26
27
28
def __init__(self, data_dir: Path | None = None):
    self.filepath = get_used_commitments_path(data_dir)
    self.used_commitments: set[str] = set()
    self.external_commitments: dict = {}
    self._load()
generate_fresh_commitment(wallet_utxos: list[UTXOInfo], cj_amount: int, private_key_getter: Any, min_confirmations: int = 5, min_percent: int = 20, max_retries: int = 3) -> ExtendedPoDLECommitment | None

Generate a fresh PoDLE commitment for a CoinJoin.

Iterates through eligible UTXOs and tries indices 0..max_retries-1 until finding an unused commitment. UTXOs are pre-sorted by confirmations and value, so fresh UTXOs (which succeed at index 0) are naturally preferred.

Args: wallet_utxos: Available wallet UTXOs cj_amount: CoinJoin amount private_key_getter: Function to get private key for address min_confirmations: Minimum UTXO confirmations required min_percent: Minimum UTXO value as % of cj_amount max_retries: Maximum number of retries per UTXO (default: 3)

Returns: ExtendedPoDLECommitment or None if no fresh commitment available

Source code in taker/src/taker/podle_manager.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def generate_fresh_commitment(
    self,
    wallet_utxos: list[UTXOInfo],
    cj_amount: int,
    private_key_getter: Any,  # Callable[[str], bytes]
    min_confirmations: int = 5,
    min_percent: int = 20,
    max_retries: int = 3,
) -> ExtendedPoDLECommitment | None:
    """
    Generate a fresh PoDLE commitment for a CoinJoin.

    Iterates through eligible UTXOs and tries indices 0..max_retries-1 until
    finding an unused commitment. UTXOs are pre-sorted by confirmations and value,
    so fresh UTXOs (which succeed at index 0) are naturally preferred.

    Args:
        wallet_utxos: Available wallet UTXOs
        cj_amount: CoinJoin amount
        private_key_getter: Function to get private key for address
        min_confirmations: Minimum UTXO confirmations required
        min_percent: Minimum UTXO value as % of cj_amount
        max_retries: Maximum number of retries per UTXO (default: 3)

    Returns:
        ExtendedPoDLECommitment or None if no fresh commitment available
    """
    eligible_utxos = get_eligible_podle_utxos(
        wallet_utxos, cj_amount, min_confirmations, min_percent
    )

    if not eligible_utxos:
        logger.warning("No eligible UTXOs for PoDLE")
        return None

    # Try each UTXO in order (already sorted by confirmations, value)
    # Fresh UTXOs naturally succeed faster (at index 0)
    for utxo in eligible_utxos:
        private_key = private_key_getter(utxo.address)
        if private_key is None:
            continue

        utxo_str = f"{utxo.txid}:{utxo.vout}"

        # Try indices 0..max_retries-1 for this UTXO
        for index in range(max_retries):
            try:
                # Generate commitment to check hash
                podle = generate_podle(private_key, utxo_str, index)
                commitment_hex = podle.commitment.hex()

                if commitment_hex in self.used_commitments:
                    logger.debug(f"PoDLE commitment for {utxo_str} index {index} already used")
                    continue

                # Found unused commitment
                self.used_commitments.add(commitment_hex)
                self._save()

                logger.info(
                    f"Generated fresh PoDLE for {utxo_str} using index {index} "
                    f"(utxo value={utxo.value}, confs={utxo.confirmations})"
                )

                return ExtendedPoDLECommitment(
                    commitment=podle,
                    scriptpubkey=utxo.scriptpubkey,
                    blockheight=utxo.height,
                )
            except Exception as e:
                logger.warning(f"Failed to generate PoDLE for {utxo_str} index {index}: {e}")
                continue

        # All indices exhausted for this UTXO
        logger.debug(f"Skipping {utxo.txid}:{utxo.vout} - all {max_retries} indices used")

    logger.error("Failed to generate any fresh PoDLE commitment from available UTXOs")
    return None
get_utxo_retry_count(utxo_str: str, private_key: bytes, max_retries: int) -> int

Get the number of times a UTXO has been used for PoDLE commitments.

Checks indices 0..(max_retries-1) in reverse order and returns the highest index + 1 where a commitment is found in used_commitments.

Note: Only used in tests. Production code uses lazy evaluation in generate_fresh_commitment() to avoid generating all commitments upfront.

Returns: 0 if UTXO is fresh (no used commitments) 1-max_retries if UTXO has been used that many times

Source code in taker/src/taker/podle_manager.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_utxo_retry_count(self, utxo_str: str, private_key: bytes, max_retries: int) -> int:
    """
    Get the number of times a UTXO has been used for PoDLE commitments.

    Checks indices 0..(max_retries-1) in reverse order and returns the highest
    index + 1 where a commitment is found in used_commitments.

    Note: Only used in tests. Production code uses lazy evaluation in
    generate_fresh_commitment() to avoid generating all commitments upfront.

    Returns:
        0 if UTXO is fresh (no used commitments)
        1-max_retries if UTXO has been used that many times
    """
    # Early termination: stop at first match (reverse order)
    for i in reversed(range(max_retries)):
        try:
            podle = generate_podle(private_key, utxo_str, i)
            commitment_hex = podle.commitment.hex()
            if commitment_hex in self.used_commitments:
                return i + 1  # Found highest used index
        except Exception:
            continue
    return 0  # No used commitments found

Functions