jmwallet.wallet.utxo_metadata

jmwallet.wallet.utxo_metadata

UTXO and address metadata persistence using BIP-329 wallet labels export format.

Stores UTXO-level metadata (frozen state, labels) and address-level metadata (addresses with on-chain history) in a single JSONL file. Each line is a BIP-329 record. This enables interoperability with external wallets like Sparrow for coin control and labeling.

BIP-329 format (JSON Lines)::

{"type": "output", "ref": "txid:vout", "spendable": false}
{"type": "output", "ref": "txid:vout", "label": "cold storage"}
{"type": "addr",   "ref": "<address>", "label": "jm:used:deposit"}

The spendable field maps to frozen state: - spendable: false -> UTXO is frozen - spendable: true or absent -> UTXO is spendable (not frozen)

The addr records track which on-chain addresses the wallet has ever held funds at (including spent-then-empty addresses). This is a privacy-critical guarantee: once an address has been observed with any UTXO it must never be reissued as a "next unused" deposit address. Light-client backends (Neutrino) and Bitcoin Core's address-book-bound RPCs alone cannot give us that guarantee across restarts; the persistent addr records do.

Label convention (informational, ignored by other BIP-329 consumers): jm:used[:<origin>] where origin is one of deposit, change, cj_out, cj_in, send (or a comma-separated combination). The origin part is best-effort context; the mere presence of the record is the privacy-relevant fact.

Reference: https://github.com/bitcoin/bips/blob/master/bip-0329.mediawiki

Attributes

DEFAULT_COINJOIN_LOCK_TTL = 600.0 module-attribute

USED_LABEL_PREFIX = 'jm:used' module-attribute

Classes

AddressRecord dataclass

A BIP-329 addr record marking an address with on-chain history.

The mere presence of a record means: this address has been observed holding (or having held) funds and must never be reissued. The label encodes optional origin context using the jm:used[:origin] convention.

Attributes: ref: Bitcoin address. label: jm:used or jm:used:<origin> (deposit, change, cj_out, cj_in, send).

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

@dataclass
class AddressRecord:
    """A BIP-329 ``addr`` record marking an address with on-chain history.

    The mere presence of a record means: this address has been observed
    holding (or having held) funds and must never be reissued. The ``label``
    encodes optional origin context using the ``jm:used[:origin]`` convention.

    Attributes:
        ref: Bitcoin address.
        label: ``jm:used`` or ``jm:used:<origin>`` (``deposit``, ``change``,
            ``cj_out``, ``cj_in``, ``send``).
    """

    ref: str
    label: str = USED_LABEL_PREFIX

    @property
    def origins(self) -> set[str]:
        """Decode the comma-separated origin set from the label, if any."""
        if not self.label.startswith(USED_LABEL_PREFIX):
            return set()
        rest = self.label[len(USED_LABEL_PREFIX) :]
        if not rest.startswith(":"):
            return set()
        return {part.strip() for part in rest[1:].split(",") if part.strip()}

    def with_added_origin(self, origin: str | None) -> AddressRecord:
        """Return a copy of this record with ``origin`` merged into the label."""
        if origin is None:
            return self
        origins = self.origins
        if origin in origins:
            return self
        origins.add(origin)
        new_label = f"{USED_LABEL_PREFIX}:{','.join(sorted(origins))}"
        return AddressRecord(ref=self.ref, label=new_label)

    def to_dict(self) -> dict[str, str]:
        """Serialize to a BIP-329 JSON dict."""
        return {"type": "addr", "ref": self.ref, "label": self.label}

    @classmethod
    def from_dict(cls, d: dict[str, str | bool]) -> AddressRecord | None:
        """Deserialize from a BIP-329 JSON dict.

        Returns ``None`` unless this is a ``type=addr`` record bearing our
        ``jm:used`` label convention; addr records labeled by other tools
        (Sparrow user labels etc.) are not treated as used-address markers
        and are preserved verbatim by ``UTXOMetadataStore``.
        """
        if d.get("type") != "addr":
            return None
        ref = d.get("ref")
        label = d.get("label", USED_LABEL_PREFIX)
        if not isinstance(ref, str) or not ref:
            return None
        if not isinstance(label, str):
            return None
        if not label.startswith(USED_LABEL_PREFIX):
            return None
        return cls(ref=ref, label=label)

Attributes

label: str = USED_LABEL_PREFIX class-attribute instance-attribute

origins: set[str] property

Decode the comma-separated origin set from the label, if any.

ref: str instance-attribute

Functions

from_dict(d: dict[str, str | bool]) -> AddressRecord | None classmethod

Deserialize from a BIP-329 JSON dict.

Returns None unless this is a type=addr record bearing our jm:used label convention; addr records labeled by other tools (Sparrow user labels etc.) are not treated as used-address markers and are preserved verbatim by UTXOMetadataStore.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

@classmethod
def from_dict(cls, d: dict[str, str | bool]) -> AddressRecord | None:
    """Deserialize from a BIP-329 JSON dict.

    Returns ``None`` unless this is a ``type=addr`` record bearing our
    ``jm:used`` label convention; addr records labeled by other tools
    (Sparrow user labels etc.) are not treated as used-address markers
    and are preserved verbatim by ``UTXOMetadataStore``.
    """
    if d.get("type") != "addr":
        return None
    ref = d.get("ref")
    label = d.get("label", USED_LABEL_PREFIX)
    if not isinstance(ref, str) or not ref:
        return None
    if not isinstance(label, str):
        return None
    if not label.startswith(USED_LABEL_PREFIX):
        return None
    return cls(ref=ref, label=label)

to_dict() -> dict[str, str]

Serialize to a BIP-329 JSON dict.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def to_dict(self) -> dict[str, str]:
    """Serialize to a BIP-329 JSON dict."""
    return {"type": "addr", "ref": self.ref, "label": self.label}

with_added_origin(origin: str | None) -> AddressRecord

Return a copy of this record with origin merged into the label.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def with_added_origin(self, origin: str | None) -> AddressRecord:
    """Return a copy of this record with ``origin`` merged into the label."""
    if origin is None:
        return self
    origins = self.origins
    if origin in origins:
        return self
    origins.add(origin)
    new_label = f"{USED_LABEL_PREFIX}:{','.join(sorted(origins))}"
    return AddressRecord(ref=self.ref, label=new_label)

OutputRecord dataclass

A BIP-329 output record for UTXO metadata.

Attributes: ref: Outpoint string in txid:vout format. spendable: Whether the UTXO is spendable. False means frozen. None means no opinion (importing wallet should not alter state). label: Optional human-readable label.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

@dataclass
class OutputRecord:
    """A BIP-329 output record for UTXO metadata.

    Attributes:
        ref: Outpoint string in ``txid:vout`` format.
        spendable: Whether the UTXO is spendable. ``False`` means frozen.
            ``None`` means no opinion (importing wallet should not alter state).
        label: Optional human-readable label.
    """

    ref: str
    spendable: bool | None = None
    label: str | None = None
    lock_until: float | None = None

    @property
    def is_frozen(self) -> bool:
        """Whether this UTXO is frozen (not spendable)."""
        return self.spendable is False

    def is_locked(self, now: float) -> bool:
        """Whether this UTXO holds a non-expired temporary CoinJoin lock.

        A lock is a *time-limited* reservation (distinct from a user freeze):
        it is set while an input is committed to an in-flight CoinJoin so that
        another concurrent round (in this or another process, maker or taker)
        does not select the same UTXO and create a conflicting transaction. It
        auto-expires after ``lock_until`` so a crashed/killed round never blocks
        funds forever.
        """
        return self.lock_until is not None and self.lock_until > now

    @property
    def has_metadata(self) -> bool:
        """Whether this record carries any state worth persisting."""
        return self.spendable is not None or self.label is not None or self.lock_until is not None

    def to_dict(self) -> dict[str, str | bool | float]:
        """Serialize to a BIP-329 JSON dict.

        ``jm_lock_until`` is a JoinMarket extension (other BIP-329 consumers
        ignore unknown keys); it carries the temporary CoinJoin lock expiry.
        """
        d: dict[str, str | bool | float] = {"type": "output", "ref": self.ref}
        if self.spendable is not None:
            d["spendable"] = self.spendable
        if self.label is not None:
            d["label"] = self.label
        if self.lock_until is not None:
            d["jm_lock_until"] = self.lock_until
        return d

    @classmethod
    def from_dict(cls, d: dict[str, str | bool | float]) -> OutputRecord | None:
        """Deserialize from a BIP-329 JSON dict.

        Returns None if the record is not a valid output record.
        """
        if d.get("type") != "output":
            return None
        ref = d.get("ref")
        if not isinstance(ref, str):
            return None
        spendable = d.get("spendable")
        if spendable is not None and not isinstance(spendable, bool):
            return None
        label = d.get("label")
        if label is not None and not isinstance(label, str):
            label = str(label)
        lock_until_raw = d.get("jm_lock_until")
        lock_until: float | None
        if isinstance(lock_until_raw, (int, float)) and not isinstance(lock_until_raw, bool):
            lock_until = float(lock_until_raw)
        else:
            lock_until = None
        return cls(ref=ref, spendable=spendable, label=label, lock_until=lock_until)

Attributes

has_metadata: bool property

Whether this record carries any state worth persisting.

is_frozen: bool property

Whether this UTXO is frozen (not spendable).

label: str | None = None class-attribute instance-attribute

lock_until: float | None = None class-attribute instance-attribute

ref: str instance-attribute

spendable: bool | None = None class-attribute instance-attribute

Functions

from_dict(d: dict[str, str | bool | float]) -> OutputRecord | None classmethod

Deserialize from a BIP-329 JSON dict.

Returns None if the record is not a valid output record.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

@classmethod
def from_dict(cls, d: dict[str, str | bool | float]) -> OutputRecord | None:
    """Deserialize from a BIP-329 JSON dict.

    Returns None if the record is not a valid output record.
    """
    if d.get("type") != "output":
        return None
    ref = d.get("ref")
    if not isinstance(ref, str):
        return None
    spendable = d.get("spendable")
    if spendable is not None and not isinstance(spendable, bool):
        return None
    label = d.get("label")
    if label is not None and not isinstance(label, str):
        label = str(label)
    lock_until_raw = d.get("jm_lock_until")
    lock_until: float | None
    if isinstance(lock_until_raw, (int, float)) and not isinstance(lock_until_raw, bool):
        lock_until = float(lock_until_raw)
    else:
        lock_until = None
    return cls(ref=ref, spendable=spendable, label=label, lock_until=lock_until)

is_locked(now: float) -> bool

Whether this UTXO holds a non-expired temporary CoinJoin lock.

A lock is a time-limited reservation (distinct from a user freeze): it is set while an input is committed to an in-flight CoinJoin so that another concurrent round (in this or another process, maker or taker) does not select the same UTXO and create a conflicting transaction. It auto-expires after lock_until so a crashed/killed round never blocks funds forever.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def is_locked(self, now: float) -> bool:
    """Whether this UTXO holds a non-expired temporary CoinJoin lock.

    A lock is a *time-limited* reservation (distinct from a user freeze):
    it is set while an input is committed to an in-flight CoinJoin so that
    another concurrent round (in this or another process, maker or taker)
    does not select the same UTXO and create a conflicting transaction. It
    auto-expires after ``lock_until`` so a crashed/killed round never blocks
    funds forever.
    """
    return self.lock_until is not None and self.lock_until > now

to_dict() -> dict[str, str | bool | float]

Serialize to a BIP-329 JSON dict.

jm_lock_until is a JoinMarket extension (other BIP-329 consumers ignore unknown keys); it carries the temporary CoinJoin lock expiry.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def to_dict(self) -> dict[str, str | bool | float]:
    """Serialize to a BIP-329 JSON dict.

    ``jm_lock_until`` is a JoinMarket extension (other BIP-329 consumers
    ignore unknown keys); it carries the temporary CoinJoin lock expiry.
    """
    d: dict[str, str | bool | float] = {"type": "output", "ref": self.ref}
    if self.spendable is not None:
        d["spendable"] = self.spendable
    if self.label is not None:
        d["label"] = self.label
    if self.lock_until is not None:
        d["jm_lock_until"] = self.lock_until
    return d

UTXOMetadataStore dataclass

In-memory store for UTXO + address metadata backed by a BIP-329 JSONL file.

Thread-safety: This class is NOT thread-safe. If concurrent access is needed, external synchronization must be applied.

Attributes: path: Path to the JSONL file on disk. records: Mapping from outpoint (txid:vout) to OutputRecord. address_records: Mapping from address to AddressRecord (only those we own with our jm:used label convention). foreign_addr_lines: Verbatim BIP-329 addr records written by other tools (Sparrow user labels, etc.). Preserved on save so we do not silently drop interoperable metadata we did not create.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

@dataclass
class UTXOMetadataStore:
    """In-memory store for UTXO + address metadata backed by a BIP-329 JSONL file.

    Thread-safety: This class is NOT thread-safe. If concurrent access is
    needed, external synchronization must be applied.

    Attributes:
        path: Path to the JSONL file on disk.
        records: Mapping from outpoint (``txid:vout``) to ``OutputRecord``.
        address_records: Mapping from address to ``AddressRecord`` (only those
            we own with our ``jm:used`` label convention).
        foreign_addr_lines: Verbatim BIP-329 ``addr`` records written by
            other tools (Sparrow user labels, etc.). Preserved on save so we
            do not silently drop interoperable metadata we did not create.
    """

    path: Path
    records: dict[str, OutputRecord] = field(default_factory=dict)
    address_records: dict[str, AddressRecord] = field(default_factory=dict)
    foreign_addr_lines: list[dict[str, str | bool]] = field(default_factory=list)

    def load(self) -> None:
        """Load metadata from disk.

        Gracefully handles missing files, empty files, and malformed lines.
        Lines that cannot be parsed are logged and skipped.
        """
        self.records.clear()
        self.address_records.clear()
        self.foreign_addr_lines.clear()

        if not self.path.exists():
            logger.debug(f"No wallet metadata file at {self.path}")
            return

        try:
            text = self.path.read_text(encoding="utf-8")
        except OSError as e:
            logger.error(f"Failed to read wallet metadata: {e}")
            return

        for line_no, line in enumerate(text.splitlines(), start=1):
            line = line.strip()
            if not line:
                continue
            try:
                data = json.loads(line)
            except json.JSONDecodeError as e:
                logger.warning(f"Malformed JSON at {self.path}:{line_no}: {e}")
                continue

            record_type = data.get("type") if isinstance(data, dict) else None
            if record_type == "output":
                record = OutputRecord.from_dict(data)
                if record is not None:
                    self.records[record.ref] = record
            elif record_type == "addr":
                # Only our ``jm:used`` labels count as used-address markers.
                # Foreign addr records (Sparrow address-book labels etc.) are
                # preserved verbatim so we round-trip third-party metadata.
                label = data.get("label")
                if isinstance(label, str) and label.startswith(USED_LABEL_PREFIX):
                    rec = AddressRecord.from_dict(data)
                    if rec is not None:
                        self.address_records[rec.ref] = rec
                else:
                    if isinstance(data, dict):
                        self.foreign_addr_lines.append(data)
            else:
                # BIP-329 says ignore unknown types -- but preserve them so we
                # do not silently drop interoperable data.
                if isinstance(data, dict):
                    self.foreign_addr_lines.append(data)

        frozen_count = sum(1 for r in self.records.values() if r.is_frozen)
        if self.records or self.address_records:
            # Make it clear these counts come from the on-disk persisted
            # state (BIP-329 metadata), not from the current bitcoind
            # sync. A wallet that has been used in the past can carry a
            # nonzero "previously used" address count even when the
            # current node returns zero history (e.g. transient RPC
            # failure, pruned data, mid-rescan), and that is intentional:
            # the persisted store is monotonic so we never re-propose a
            # deposit address that was historically funded.
            logger.debug(
                f"Loaded persisted wallet metadata from {self.path}: "
                f"{len(self.records)} UTXO record(s) ({frozen_count} frozen), "
                f"{len(self.address_records)} previously used address(es), "
                f"and {len(self.foreign_addr_lines)} foreign record(s). "
                f"These are read from disk and reflect historical state, "
                f"not the result of the current bitcoind scan."
            )

    def save(self) -> None:
        """Persist all records to disk.

        Writes the entire file atomically (write to temp, then rename) to
        prevent corruption on crash. ``output`` records, our ``jm:used``
        ``addr`` records, and any foreign records loaded from disk are all
        serialized in a deterministic order.

        Raises:
            OSError: If the file cannot be written (e.g., read-only filesystem).
        """
        self.path.parent.mkdir(parents=True, exist_ok=True)

        # Filter out output records that carry no useful metadata
        outputs_to_write = [r for r in self.records.values() if r.has_metadata]
        outputs_to_write.sort(key=lambda r: r.ref)

        addr_records_to_write = sorted(self.address_records.values(), key=lambda r: r.ref)

        if not outputs_to_write and not addr_records_to_write and not self.foreign_addr_lines:
            if self.path.exists():
                try:
                    self.path.unlink()
                    logger.debug("Removed empty wallet metadata file")
                except OSError as e:
                    logger.warning(f"Failed to remove empty metadata file: {e}")
                    raise
            return

        lines: list[str] = []
        lines.extend(json.dumps(r.to_dict(), separators=(",", ":")) for r in outputs_to_write)
        lines.extend(json.dumps(r.to_dict(), separators=(",", ":")) for r in addr_records_to_write)
        # Foreign records last; sort by (type, ref) for determinism.
        for foreign in sorted(
            self.foreign_addr_lines,
            key=lambda d: (str(d.get("type", "")), str(d.get("ref", ""))),
        ):
            lines.append(json.dumps(foreign, separators=(",", ":")))

        tmp_path = self.path.with_suffix(".tmp")
        try:
            tmp_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
            tmp_path.replace(self.path)
        except OSError as e:
            logger.error(f"Failed to save wallet metadata: {e}")
            try:
                tmp_path.unlink(missing_ok=True)
            except OSError:
                pass
            raise

    def is_frozen(self, outpoint: str) -> bool:
        """Check if an outpoint is frozen.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Returns:
            True if the UTXO is frozen (spendable is False).
        """
        record = self.records.get(outpoint)
        return record is not None and record.is_frozen

    def get_frozen_outpoints(self) -> set[str]:
        """Get all frozen outpoints.

        Returns:
            Set of outpoint strings that are frozen.
        """
        return {ref for ref, record in self.records.items() if record.is_frozen}

    def freeze(self, outpoint: str) -> None:
        """Freeze a UTXO (set spendable to False) and persist.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.
        """
        if outpoint in self.records:
            self.records[outpoint].spendable = False
        else:
            self.records[outpoint] = OutputRecord(ref=outpoint, spendable=False)
        self.save()
        logger.info(f"Frozen UTXO: {outpoint}")

    def unfreeze(self, outpoint: str) -> None:
        """Unfreeze a UTXO (set spendable to True) and persist.

        If the record has no other metadata (no label), it is removed
        entirely since ``spendable=True`` is the default.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.
        """
        record = self.records.get(outpoint)
        if record is None:
            # Already unfrozen (no record means spendable)
            return

        if record.label is not None or record.lock_until is not None:
            # Keep the record for the label / active CoinJoin lock.
            record.spendable = True
        else:
            # No other metadata -- remove entirely
            del self.records[outpoint]

        self.save()
        logger.info(f"Unfrozen UTXO: {outpoint}")

    def toggle_freeze(self, outpoint: str) -> bool:
        """Toggle the frozen state of a UTXO and persist.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Returns:
            True if the UTXO is now frozen, False if now unfrozen.
        """
        if self.is_frozen(outpoint):
            self.unfreeze(outpoint)
            return False
        else:
            self.freeze(outpoint)
            return True

    # -- Temporary CoinJoin locks --------------------------------------------
    #
    # A lock is a *time-limited* reservation on an input that is committed to an
    # in-flight CoinJoin. It is persisted in the same JSONL file (via
    # ``jm_lock_until``) so that other processes -- another taker round, or a
    # maker serving a different taker -- re-read it right before coin selection
    # and never pick the same UTXO. Picking the same input twice produces
    # conflicting, mutually double-spending transactions; the one broadcast
    # second is rejected ("insufficient fee, rejecting replacement"). Locks
    # auto-expire (``lock_until``) so a crashed or killed round cannot block
    # funds forever, and acquisition is serialized across processes with an
    # advisory file lock so two processes cannot both win the same UTXO.

    @property
    def _flock_path(self) -> Path:
        return self.path.with_suffix(".lock")

    @contextmanager
    def _exclusive_file_lock(self) -> Iterator[None]:
        """Serialize lock acquisition/release across processes.

        Uses a POSIX advisory lock on a sidecar ``.lock`` file. The metadata
        file itself is replaced via rename on every save, which would break a
        lock held on its inode, so we lock a stable sidecar path instead. On
        platforms without ``fcntl`` this degrades to a no-op (atomic rename
        still prevents corruption; only the rare lost-update race remains).
        """
        if fcntl is None:  # pragma: no cover - non-POSIX platforms
            yield
            return
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with open(self._flock_path, "w", encoding="utf-8") as handle:
            fcntl.flock(handle, fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle, fcntl.LOCK_UN)

    def _prune_expired_locks(self, now: float) -> None:
        """Clear expired ``lock_until`` markers and drop now-empty records."""
        for ref in list(self.records.keys()):
            record = self.records[ref]
            if record.lock_until is not None and record.lock_until <= now:
                record.lock_until = None
                if not record.has_metadata:
                    del self.records[ref]

    def get_locked_outpoints(self, now: float | None = None) -> set[str]:
        """Return outpoints currently holding a non-expired CoinJoin lock.

        Note: reads in-memory state. Call :meth:`load` first to observe locks
        written by other processes.
        """
        if now is None:
            now = time.time()
        return {ref for ref, record in self.records.items() if record.is_locked(now)}

    def try_lock_outpoints(
        self, outpoints: Iterable[str], ttl: float = DEFAULT_COINJOIN_LOCK_TTL
    ) -> bool:
        """Atomically lock ``outpoints`` for ``ttl`` seconds.

        Reloads on-disk state under an exclusive file lock so concurrent
        processes cannot both acquire the same UTXO. Fails (returning False,
        locking nothing) if any requested outpoint is frozen or already locked
        by another in-flight round.

        Returns:
            True if all outpoints were locked; False on conflict.
        """
        wanted = set(outpoints)
        if not wanted:
            return True
        with self._exclusive_file_lock():
            self.load()
            now = time.time()
            self._prune_expired_locks(now)
            for ref in wanted:
                record = self.records.get(ref)
                if record is not None and (record.is_frozen or record.is_locked(now)):
                    return False
            for ref in wanted:
                record = self.records.get(ref)
                if record is None:
                    record = OutputRecord(ref=ref)
                    self.records[ref] = record
                record.lock_until = now + ttl
            self.save()
        return True

    def release_outpoints(self, outpoints: Iterable[str]) -> None:
        """Clear CoinJoin locks on ``outpoints`` (no-op for unlocked ones)."""
        wanted = set(outpoints)
        if not wanted:
            return
        with self._exclusive_file_lock():
            self.load()
            changed = False
            for ref in wanted:
                record = self.records.get(ref)
                if record is not None and record.lock_until is not None:
                    record.lock_until = None
                    changed = True
                    if not record.has_metadata:
                        del self.records[ref]
            self._prune_expired_locks(time.time())
            if changed:
                self.save()

    def set_label(self, outpoint: str, label: str | None) -> None:
        """Set or clear the label for a UTXO and persist.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.
            label: Label string, or None to clear.
        """
        if outpoint in self.records:
            self.records[outpoint].label = label
        elif label is not None:
            self.records[outpoint] = OutputRecord(ref=outpoint, label=label)
        else:
            return  # Nothing to do

        # Clean up record if it has no useful metadata
        record = self.records.get(outpoint)
        if record and record.spendable is None and record.label is None:
            del self.records[outpoint]

        self.save()

    def get_label(self, outpoint: str) -> str | None:
        """Get the label for an outpoint.

        Args:
            outpoint: Outpoint string in ``txid:vout`` format.

        Returns:
            Label string, or None if no label set.
        """
        record = self.records.get(outpoint)
        return record.label if record else None

    # -- Address history (BIP-329 ``addr`` records with ``jm:used`` label) --

    def mark_address_used(self, address: str, origin: str | None = None) -> bool:
        """Record an address as having on-chain history.

        Idempotent. If the address is already recorded, only the origin label
        is augmented (best-effort context); the file is rewritten only when
        the record actually changes. Returns ``True`` if disk state changed.
        """
        if not address:
            return False
        existing = self.address_records.get(address)
        if existing is None:
            self.address_records[address] = AddressRecord(
                ref=address,
                label=f"{USED_LABEL_PREFIX}:{origin}" if origin else USED_LABEL_PREFIX,
            )
            self.save()
            return True
        updated = existing.with_added_origin(origin)
        if updated.label == existing.label:
            return False
        self.address_records[address] = updated
        self.save()
        return True

    def mark_addresses_used(
        self,
        addresses: Iterable[str],
        origin: str | None = None,
    ) -> int:
        """Batched variant of :meth:`mark_address_used`.

        Performs a single ``save()`` for many addresses; returns the count of
        records that were created or had their origin extended.
        """
        changed = 0
        for address in addresses:
            if not address:
                continue
            existing = self.address_records.get(address)
            if existing is None:
                self.address_records[address] = AddressRecord(
                    ref=address,
                    label=f"{USED_LABEL_PREFIX}:{origin}" if origin else USED_LABEL_PREFIX,
                )
                changed += 1
                continue
            updated = existing.with_added_origin(origin)
            if updated.label != existing.label:
                self.address_records[address] = updated
                changed += 1
        if changed:
            self.save()
        return changed

    def is_address_used(self, address: str) -> bool:
        """Return True if ``address`` has been recorded as having history."""
        return address in self.address_records

    def get_used_addresses(self) -> set[str]:
        """Return the set of addresses with on-chain history.

        This is the privacy-critical "do not reissue" set, surviving across
        process restarts and backend swaps.
        """
        return set(self.address_records.keys())

    def verify_writable(self) -> None:
        """Verify that the metadata file's directory is writable.

        Attempts to create and immediately remove a temporary file in the
        same directory as the metadata file. This catches read-only mounts
        and permission issues early, before a real save attempt.

        Raises:
            OSError: If the directory is not writable.
        """
        parent = self.path.parent
        parent.mkdir(parents=True, exist_ok=True)
        # Try creating a temp file in the target directory
        try:
            fd = tempfile.NamedTemporaryFile(dir=parent, prefix=".jm_write_test_", delete=True)
            fd.close()
        except OSError as e:
            raise OSError(
                f"Data directory is not writable: {parent}. "
                f"Cannot persist UTXO metadata (frozen state, labels). "
                f"Check mount permissions. Original error: {e}"
            ) from e

Attributes

address_records: dict[str, AddressRecord] = field(default_factory=dict) class-attribute instance-attribute

foreign_addr_lines: list[dict[str, str | bool]] = field(default_factory=list) class-attribute instance-attribute

path: Path instance-attribute

records: dict[str, OutputRecord] = field(default_factory=dict) class-attribute instance-attribute

Functions

freeze(outpoint: str) -> None

Freeze a UTXO (set spendable to False) and persist.

Args: outpoint: Outpoint string in txid:vout format.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def freeze(self, outpoint: str) -> None:
    """Freeze a UTXO (set spendable to False) and persist.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.
    """
    if outpoint in self.records:
        self.records[outpoint].spendable = False
    else:
        self.records[outpoint] = OutputRecord(ref=outpoint, spendable=False)
    self.save()
    logger.info(f"Frozen UTXO: {outpoint}")

get_frozen_outpoints() -> set[str]

Get all frozen outpoints.

Returns: Set of outpoint strings that are frozen.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def get_frozen_outpoints(self) -> set[str]:
    """Get all frozen outpoints.

    Returns:
        Set of outpoint strings that are frozen.
    """
    return {ref for ref, record in self.records.items() if record.is_frozen}

get_label(outpoint: str) -> str | None

Get the label for an outpoint.

Args: outpoint: Outpoint string in txid:vout format.

Returns: Label string, or None if no label set.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def get_label(self, outpoint: str) -> str | None:
    """Get the label for an outpoint.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Returns:
        Label string, or None if no label set.
    """
    record = self.records.get(outpoint)
    return record.label if record else None

get_locked_outpoints(now: float | None = None) -> set[str]

Return outpoints currently holding a non-expired CoinJoin lock.

Note: reads in-memory state. Call :meth:load first to observe locks written by other processes.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def get_locked_outpoints(self, now: float | None = None) -> set[str]:
    """Return outpoints currently holding a non-expired CoinJoin lock.

    Note: reads in-memory state. Call :meth:`load` first to observe locks
    written by other processes.
    """
    if now is None:
        now = time.time()
    return {ref for ref, record in self.records.items() if record.is_locked(now)}

get_used_addresses() -> set[str]

Return the set of addresses with on-chain history.

This is the privacy-critical "do not reissue" set, surviving across process restarts and backend swaps.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def get_used_addresses(self) -> set[str]:
    """Return the set of addresses with on-chain history.

    This is the privacy-critical "do not reissue" set, surviving across
    process restarts and backend swaps.
    """
    return set(self.address_records.keys())

is_address_used(address: str) -> bool

Return True if address has been recorded as having history.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def is_address_used(self, address: str) -> bool:
    """Return True if ``address`` has been recorded as having history."""
    return address in self.address_records

is_frozen(outpoint: str) -> bool

Check if an outpoint is frozen.

Args: outpoint: Outpoint string in txid:vout format.

Returns: True if the UTXO is frozen (spendable is False).

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def is_frozen(self, outpoint: str) -> bool:
    """Check if an outpoint is frozen.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Returns:
        True if the UTXO is frozen (spendable is False).
    """
    record = self.records.get(outpoint)
    return record is not None and record.is_frozen

load() -> None

Load metadata from disk.

Gracefully handles missing files, empty files, and malformed lines. Lines that cannot be parsed are logged and skipped.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def load(self) -> None:
    """Load metadata from disk.

    Gracefully handles missing files, empty files, and malformed lines.
    Lines that cannot be parsed are logged and skipped.
    """
    self.records.clear()
    self.address_records.clear()
    self.foreign_addr_lines.clear()

    if not self.path.exists():
        logger.debug(f"No wallet metadata file at {self.path}")
        return

    try:
        text = self.path.read_text(encoding="utf-8")
    except OSError as e:
        logger.error(f"Failed to read wallet metadata: {e}")
        return

    for line_no, line in enumerate(text.splitlines(), start=1):
        line = line.strip()
        if not line:
            continue
        try:
            data = json.loads(line)
        except json.JSONDecodeError as e:
            logger.warning(f"Malformed JSON at {self.path}:{line_no}: {e}")
            continue

        record_type = data.get("type") if isinstance(data, dict) else None
        if record_type == "output":
            record = OutputRecord.from_dict(data)
            if record is not None:
                self.records[record.ref] = record
        elif record_type == "addr":
            # Only our ``jm:used`` labels count as used-address markers.
            # Foreign addr records (Sparrow address-book labels etc.) are
            # preserved verbatim so we round-trip third-party metadata.
            label = data.get("label")
            if isinstance(label, str) and label.startswith(USED_LABEL_PREFIX):
                rec = AddressRecord.from_dict(data)
                if rec is not None:
                    self.address_records[rec.ref] = rec
            else:
                if isinstance(data, dict):
                    self.foreign_addr_lines.append(data)
        else:
            # BIP-329 says ignore unknown types -- but preserve them so we
            # do not silently drop interoperable data.
            if isinstance(data, dict):
                self.foreign_addr_lines.append(data)

    frozen_count = sum(1 for r in self.records.values() if r.is_frozen)
    if self.records or self.address_records:
        # Make it clear these counts come from the on-disk persisted
        # state (BIP-329 metadata), not from the current bitcoind
        # sync. A wallet that has been used in the past can carry a
        # nonzero "previously used" address count even when the
        # current node returns zero history (e.g. transient RPC
        # failure, pruned data, mid-rescan), and that is intentional:
        # the persisted store is monotonic so we never re-propose a
        # deposit address that was historically funded.
        logger.debug(
            f"Loaded persisted wallet metadata from {self.path}: "
            f"{len(self.records)} UTXO record(s) ({frozen_count} frozen), "
            f"{len(self.address_records)} previously used address(es), "
            f"and {len(self.foreign_addr_lines)} foreign record(s). "
            f"These are read from disk and reflect historical state, "
            f"not the result of the current bitcoind scan."
        )

mark_address_used(address: str, origin: str | None = None) -> bool

Record an address as having on-chain history.

Idempotent. If the address is already recorded, only the origin label is augmented (best-effort context); the file is rewritten only when the record actually changes. Returns True if disk state changed.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def mark_address_used(self, address: str, origin: str | None = None) -> bool:
    """Record an address as having on-chain history.

    Idempotent. If the address is already recorded, only the origin label
    is augmented (best-effort context); the file is rewritten only when
    the record actually changes. Returns ``True`` if disk state changed.
    """
    if not address:
        return False
    existing = self.address_records.get(address)
    if existing is None:
        self.address_records[address] = AddressRecord(
            ref=address,
            label=f"{USED_LABEL_PREFIX}:{origin}" if origin else USED_LABEL_PREFIX,
        )
        self.save()
        return True
    updated = existing.with_added_origin(origin)
    if updated.label == existing.label:
        return False
    self.address_records[address] = updated
    self.save()
    return True

mark_addresses_used(addresses: Iterable[str], origin: str | None = None) -> int

Batched variant of :meth:mark_address_used.

Performs a single save() for many addresses; returns the count of records that were created or had their origin extended.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def mark_addresses_used(
    self,
    addresses: Iterable[str],
    origin: str | None = None,
) -> int:
    """Batched variant of :meth:`mark_address_used`.

    Performs a single ``save()`` for many addresses; returns the count of
    records that were created or had their origin extended.
    """
    changed = 0
    for address in addresses:
        if not address:
            continue
        existing = self.address_records.get(address)
        if existing is None:
            self.address_records[address] = AddressRecord(
                ref=address,
                label=f"{USED_LABEL_PREFIX}:{origin}" if origin else USED_LABEL_PREFIX,
            )
            changed += 1
            continue
        updated = existing.with_added_origin(origin)
        if updated.label != existing.label:
            self.address_records[address] = updated
            changed += 1
    if changed:
        self.save()
    return changed

release_outpoints(outpoints: Iterable[str]) -> None

Clear CoinJoin locks on outpoints (no-op for unlocked ones).

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def release_outpoints(self, outpoints: Iterable[str]) -> None:
    """Clear CoinJoin locks on ``outpoints`` (no-op for unlocked ones)."""
    wanted = set(outpoints)
    if not wanted:
        return
    with self._exclusive_file_lock():
        self.load()
        changed = False
        for ref in wanted:
            record = self.records.get(ref)
            if record is not None and record.lock_until is not None:
                record.lock_until = None
                changed = True
                if not record.has_metadata:
                    del self.records[ref]
        self._prune_expired_locks(time.time())
        if changed:
            self.save()

save() -> None

Persist all records to disk.

Writes the entire file atomically (write to temp, then rename) to prevent corruption on crash. output records, our jm:used addr records, and any foreign records loaded from disk are all serialized in a deterministic order.

Raises: OSError: If the file cannot be written (e.g., read-only filesystem).

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def save(self) -> None:
    """Persist all records to disk.

    Writes the entire file atomically (write to temp, then rename) to
    prevent corruption on crash. ``output`` records, our ``jm:used``
    ``addr`` records, and any foreign records loaded from disk are all
    serialized in a deterministic order.

    Raises:
        OSError: If the file cannot be written (e.g., read-only filesystem).
    """
    self.path.parent.mkdir(parents=True, exist_ok=True)

    # Filter out output records that carry no useful metadata
    outputs_to_write = [r for r in self.records.values() if r.has_metadata]
    outputs_to_write.sort(key=lambda r: r.ref)

    addr_records_to_write = sorted(self.address_records.values(), key=lambda r: r.ref)

    if not outputs_to_write and not addr_records_to_write and not self.foreign_addr_lines:
        if self.path.exists():
            try:
                self.path.unlink()
                logger.debug("Removed empty wallet metadata file")
            except OSError as e:
                logger.warning(f"Failed to remove empty metadata file: {e}")
                raise
        return

    lines: list[str] = []
    lines.extend(json.dumps(r.to_dict(), separators=(",", ":")) for r in outputs_to_write)
    lines.extend(json.dumps(r.to_dict(), separators=(",", ":")) for r in addr_records_to_write)
    # Foreign records last; sort by (type, ref) for determinism.
    for foreign in sorted(
        self.foreign_addr_lines,
        key=lambda d: (str(d.get("type", "")), str(d.get("ref", ""))),
    ):
        lines.append(json.dumps(foreign, separators=(",", ":")))

    tmp_path = self.path.with_suffix(".tmp")
    try:
        tmp_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
        tmp_path.replace(self.path)
    except OSError as e:
        logger.error(f"Failed to save wallet metadata: {e}")
        try:
            tmp_path.unlink(missing_ok=True)
        except OSError:
            pass
        raise

set_label(outpoint: str, label: str | None) -> None

Set or clear the label for a UTXO and persist.

Args: outpoint: Outpoint string in txid:vout format. label: Label string, or None to clear.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def set_label(self, outpoint: str, label: str | None) -> None:
    """Set or clear the label for a UTXO and persist.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.
        label: Label string, or None to clear.
    """
    if outpoint in self.records:
        self.records[outpoint].label = label
    elif label is not None:
        self.records[outpoint] = OutputRecord(ref=outpoint, label=label)
    else:
        return  # Nothing to do

    # Clean up record if it has no useful metadata
    record = self.records.get(outpoint)
    if record and record.spendable is None and record.label is None:
        del self.records[outpoint]

    self.save()

toggle_freeze(outpoint: str) -> bool

Toggle the frozen state of a UTXO and persist.

Args: outpoint: Outpoint string in txid:vout format.

Returns: True if the UTXO is now frozen, False if now unfrozen.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def toggle_freeze(self, outpoint: str) -> bool:
    """Toggle the frozen state of a UTXO and persist.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.

    Returns:
        True if the UTXO is now frozen, False if now unfrozen.
    """
    if self.is_frozen(outpoint):
        self.unfreeze(outpoint)
        return False
    else:
        self.freeze(outpoint)
        return True

try_lock_outpoints(outpoints: Iterable[str], ttl: float = DEFAULT_COINJOIN_LOCK_TTL) -> bool

Atomically lock outpoints for ttl seconds.

Reloads on-disk state under an exclusive file lock so concurrent processes cannot both acquire the same UTXO. Fails (returning False, locking nothing) if any requested outpoint is frozen or already locked by another in-flight round.

Returns: True if all outpoints were locked; False on conflict.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def try_lock_outpoints(
    self, outpoints: Iterable[str], ttl: float = DEFAULT_COINJOIN_LOCK_TTL
) -> bool:
    """Atomically lock ``outpoints`` for ``ttl`` seconds.

    Reloads on-disk state under an exclusive file lock so concurrent
    processes cannot both acquire the same UTXO. Fails (returning False,
    locking nothing) if any requested outpoint is frozen or already locked
    by another in-flight round.

    Returns:
        True if all outpoints were locked; False on conflict.
    """
    wanted = set(outpoints)
    if not wanted:
        return True
    with self._exclusive_file_lock():
        self.load()
        now = time.time()
        self._prune_expired_locks(now)
        for ref in wanted:
            record = self.records.get(ref)
            if record is not None and (record.is_frozen or record.is_locked(now)):
                return False
        for ref in wanted:
            record = self.records.get(ref)
            if record is None:
                record = OutputRecord(ref=ref)
                self.records[ref] = record
            record.lock_until = now + ttl
        self.save()
    return True

unfreeze(outpoint: str) -> None

Unfreeze a UTXO (set spendable to True) and persist.

If the record has no other metadata (no label), it is removed entirely since spendable=True is the default.

Args: outpoint: Outpoint string in txid:vout format.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def unfreeze(self, outpoint: str) -> None:
    """Unfreeze a UTXO (set spendable to True) and persist.

    If the record has no other metadata (no label), it is removed
    entirely since ``spendable=True`` is the default.

    Args:
        outpoint: Outpoint string in ``txid:vout`` format.
    """
    record = self.records.get(outpoint)
    if record is None:
        # Already unfrozen (no record means spendable)
        return

    if record.label is not None or record.lock_until is not None:
        # Keep the record for the label / active CoinJoin lock.
        record.spendable = True
    else:
        # No other metadata -- remove entirely
        del self.records[outpoint]

    self.save()
    logger.info(f"Unfrozen UTXO: {outpoint}")

verify_writable() -> None

Verify that the metadata file's directory is writable.

Attempts to create and immediately remove a temporary file in the same directory as the metadata file. This catches read-only mounts and permission issues early, before a real save attempt.

Raises: OSError: If the directory is not writable.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def verify_writable(self) -> None:
    """Verify that the metadata file's directory is writable.

    Attempts to create and immediately remove a temporary file in the
    same directory as the metadata file. This catches read-only mounts
    and permission issues early, before a real save attempt.

    Raises:
        OSError: If the directory is not writable.
    """
    parent = self.path.parent
    parent.mkdir(parents=True, exist_ok=True)
    # Try creating a temp file in the target directory
    try:
        fd = tempfile.NamedTemporaryFile(dir=parent, prefix=".jm_write_test_", delete=True)
        fd.close()
    except OSError as e:
        raise OSError(
            f"Data directory is not writable: {parent}. "
            f"Cannot persist UTXO metadata (frozen state, labels). "
            f"Check mount permissions. Original error: {e}"
        ) from e

Functions

load_metadata_store(data_dir: Path, fingerprint: str | None = None, owned_addresses: Iterable[str] | None = None) -> UTXOMetadataStore

Create and load a UTXOMetadataStore from the wallet's metadata file.

Args: data_dir: JoinMarket data directory (e.g., ~/.joinmarket-ng). fingerprint: Optional 8-char hex wallet fingerprint. When provided, the per-wallet path wallet_metadata_<fp>.jsonl is used and a one-shot migration from the legacy shared wallet_metadata.jsonl is attempted on first open. owned_addresses: Optional iterable of addresses this wallet derives inside its scan range. Used to filter addr records during migration so we do not import another wallet's used-address set from the shared file. When None no addr records are imported (safer default than "all"; the wallet's own sync will re-populate any genuinely-funded addresses).

Returns: Loaded UTXOMetadataStore instance.

Source code in jmwallet/src/jmwallet/wallet/utxo_metadata.py

def load_metadata_store(
    data_dir: Path,
    fingerprint: str | None = None,
    owned_addresses: Iterable[str] | None = None,
) -> UTXOMetadataStore:
    """Create and load a UTXOMetadataStore from the wallet's metadata file.

    Args:
        data_dir: JoinMarket data directory (e.g., ``~/.joinmarket-ng``).
        fingerprint: Optional 8-char hex wallet fingerprint. When provided,
            the per-wallet path ``wallet_metadata_<fp>.jsonl`` is used and a
            one-shot migration from the legacy shared
            ``wallet_metadata.jsonl`` is attempted on first open.
        owned_addresses: Optional iterable of addresses this wallet derives
            inside its scan range. Used to filter ``addr`` records during
            migration so we do not import another wallet's used-address set
            from the shared file. When ``None`` no ``addr`` records are
            imported (safer default than "all"; the wallet's own sync will
            re-populate any genuinely-funded addresses).

    Returns:
        Loaded UTXOMetadataStore instance.
    """
    from jmcore.paths import get_wallet_metadata_path

    path = get_wallet_metadata_path(data_dir, fingerprint=fingerprint)

    if fingerprint is not None and not path.exists():
        legacy_shared = get_wallet_metadata_path(data_dir, fingerprint=None)
        if legacy_shared.exists() and legacy_shared != path:
            _migrate_shared_metadata(
                legacy_shared,
                path,
                owned_addresses=owned_addresses,
            )

    store = UTXOMetadataStore(path=path)
    store.load()
    return store