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 
 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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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: Callable[[str], bytes | None],
        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
        """
        candidates = self._iter_fresh_commitments(
            wallet_utxos,
            cj_amount,
            private_key_getter,
            min_confirmations,
            min_percent,
            max_retries,
        )
        for utxo, podle in candidates:
            commitment_hex = podle.commitment.hex()
            self.used_commitments.add(commitment_hex)
            self._save()

            logger.info("Generated fresh PoDLE commitment")
            logger.bind(sensitive=True).info(
                "Generated fresh PoDLE for {} using index {} (utxo value={}, confs={})",
                podle.utxo,
                podle.index,
                utxo.value,
                utxo.confirmations,
            )

            return ExtendedPoDLECommitment(
                commitment=podle,
                scriptpubkey=utxo.scriptpubkey,
                blockheight=utxo.height,
            )

        logger.error("Failed to generate any fresh PoDLE commitment from available UTXOs")
        return None

    def get_fresh_commitment_utxos(
        self,
        wallet_utxos: list[UTXOInfo],
        cj_amount: int,
        private_key_getter: Callable[[str], bytes | None],
        min_confirmations: int = 5,
        min_percent: int = 20,
        max_retries: int = 3,
    ) -> list[UTXOInfo]:
        """Return PoDLE-capable UTXOs without consuming a commitment index."""
        fresh: list[UTXOInfo] = []
        seen: set[tuple[str, int]] = set()
        for utxo, _ in self._iter_fresh_commitments(
            wallet_utxos,
            cj_amount,
            private_key_getter,
            min_confirmations,
            min_percent,
            max_retries,
        ):
            outpoint = (utxo.txid, utxo.vout)
            if outpoint not in seen:
                fresh.append(utxo)
                seen.add(outpoint)
        return fresh

    def _iter_fresh_commitments(
        self,
        wallet_utxos: list[UTXOInfo],
        cj_amount: int,
        private_key_getter: Callable[[str], bytes | None],
        min_confirmations: int,
        min_percent: int,
        max_retries: int,
    ) -> Iterator[tuple[UTXOInfo, PoDLECommitment]]:
        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

        try:
            blacklist = get_blacklist()
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning(f"Could not load commitment blacklist: {exc}")
            blacklist = None

        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}"
            found = False
            for index in range(max_retries):
                try:
                    podle = generate_podle(private_key, utxo_str, index)
                    commitment_hex = podle.commitment.hex()
                    if commitment_hex in self.used_commitments:
                        logger.debug("PoDLE commitment retry index already used")
                        logger.bind(sensitive=True).debug(
                            "PoDLE commitment for {} index {} already used", utxo_str, index
                        )
                        continue
                    if blacklist is not None and blacklist.is_blacklisted(commitment_hex):
                        logger.debug("PoDLE commitment retry index is blacklisted")
                        logger.bind(sensitive=True).debug(
                            "PoDLE commitment for {} index {} is blacklisted", utxo_str, index
                        )
                        self.used_commitments.add(commitment_hex)
                        self._save()
                        continue
                    found = True
                    yield utxo, podle
                    break
                except Exception as exc:
                    logger.warning("Failed to generate PoDLE commitment")
                    logger.bind(sensitive=True).warning(
                        "Failed to generate PoDLE for {} index {}: {}", utxo_str, index, exc
                    )
            if not found:
                logger.debug("Skipping UTXO after all PoDLE retry indices were used")
                logger.bind(sensitive=True).debug(
                    "Skipping {}:{} after all {} PoDLE retry indices were used",
                    utxo.txid,
                    utxo.vout,
                    max_retries,
                )
Attributes
external_commitments: dict = {} instance-attribute
filepath = get_used_commitments_path(data_dir) instance-attribute
used_commitments: set[str] = set() instance-attribute
Methods:
__init__(data_dir: Path | None = None)
Source code in taker/src/taker/podle_manager.py 
26
27
28
29
30
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: Callable[[str], bytes | None], 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 
 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
def generate_fresh_commitment(
    self,
    wallet_utxos: list[UTXOInfo],
    cj_amount: int,
    private_key_getter: Callable[[str], bytes | None],
    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
    """
    candidates = self._iter_fresh_commitments(
        wallet_utxos,
        cj_amount,
        private_key_getter,
        min_confirmations,
        min_percent,
        max_retries,
    )
    for utxo, podle in candidates:
        commitment_hex = podle.commitment.hex()
        self.used_commitments.add(commitment_hex)
        self._save()

        logger.info("Generated fresh PoDLE commitment")
        logger.bind(sensitive=True).info(
            "Generated fresh PoDLE for {} using index {} (utxo value={}, confs={})",
            podle.utxo,
            podle.index,
            utxo.value,
            utxo.confirmations,
        )

        return ExtendedPoDLECommitment(
            commitment=podle,
            scriptpubkey=utxo.scriptpubkey,
            blockheight=utxo.height,
        )

    logger.error("Failed to generate any fresh PoDLE commitment from available UTXOs")
    return None
get_fresh_commitment_utxos(wallet_utxos: list[UTXOInfo], cj_amount: int, private_key_getter: Callable[[str], bytes | None], min_confirmations: int = 5, min_percent: int = 20, max_retries: int = 3) -> list[UTXOInfo]

Return PoDLE-capable UTXOs without consuming a commitment index.

Source code in taker/src/taker/podle_manager.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def get_fresh_commitment_utxos(
    self,
    wallet_utxos: list[UTXOInfo],
    cj_amount: int,
    private_key_getter: Callable[[str], bytes | None],
    min_confirmations: int = 5,
    min_percent: int = 20,
    max_retries: int = 3,
) -> list[UTXOInfo]:
    """Return PoDLE-capable UTXOs without consuming a commitment index."""
    fresh: list[UTXOInfo] = []
    seen: set[tuple[str, int]] = set()
    for utxo, _ in self._iter_fresh_commitments(
        wallet_utxos,
        cj_amount,
        private_key_getter,
        min_confirmations,
        min_percent,
        max_retries,
    ):
        outpoint = (utxo.txid, utxo.vout)
        if outpoint not in seen:
            fresh.append(utxo)
            seen.add(outpoint)
    return fresh
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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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: