Skip to content

jmwallet.history_state

jmwallet.history_state

Durable cursor state for incremental on-chain history reconstruction.

Classes

ReconstructionCheckpoint

Bases: BaseModel

A completed reconstruction baseline and its backend cursor.

Source code in jmwallet/src/jmwallet/history_state.py 
16
17
18
19
20
21
22
23
24
25
26
27
class ReconstructionCheckpoint(BaseModel):
    """A completed reconstruction baseline and its backend cursor."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    version: int = 1
    wallet_fingerprint: str
    network: str
    backend_id: str
    cursor: str | None = None
    regular_branch_ends: dict[str, int] = Field(default_factory=dict)
    scanned_address_hashes: set[str] = Field(default_factory=set)
Attributes
backend_id: str instance-attribute
cursor: str | None = None class-attribute instance-attribute
model_config = ConfigDict(extra='forbid', frozen=True) class-attribute instance-attribute
network: str instance-attribute
regular_branch_ends: dict[str, int] = Field(default_factory=dict) class-attribute instance-attribute
scanned_address_hashes: set[str] = Field(default_factory=set) class-attribute instance-attribute
version: int = 1 class-attribute instance-attribute
wallet_fingerprint: str instance-attribute

Functions:

clear_reconstruction_cursor(data_dir: Path, *, wallet_fingerprint: str) -> None

Invalidate incremental state before rebuilding or widening coverage.

Source code in jmwallet/src/jmwallet/history_state.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def clear_reconstruction_cursor(data_dir: Path, *, wallet_fingerprint: str) -> None:
    """Invalidate incremental state before rebuilding or widening coverage."""
    path: Path | None = None
    try:
        path = _checkpoint_path(data_dir, wallet_fingerprint)
        if not path.exists():
            return
        checkpoint = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
        checkpoint = checkpoint.model_copy(update={"cursor": None})
        atomic_write_private(path, checkpoint.model_dump_json(indent=2).encode("utf-8") + b"\n")
    except (OSError, ValidationError, ValueError) as exc:
        try:
            if path is not None:
                path.unlink(missing_ok=True)
        except OSError:
            pass
        logger.warning("Could not clear history reconstruction checkpoint")
        logger.bind(sensitive=True).warning(
            f"Could not clear history reconstruction checkpoint: {exc}"
        )

get_uncovered_reconstruction_addresses(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, addresses: Iterable[str]) -> list[str]

Return explicit addresses without durable historical-scan coverage.

Source code in jmwallet/src/jmwallet/history_state.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def get_uncovered_reconstruction_addresses(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    addresses: Iterable[str],
) -> list[str]:
    """Return explicit addresses without durable historical-scan coverage."""
    unique_addresses = list(dict.fromkeys(address.strip().lower() for address in addresses))
    try:
        checkpoint = ReconstructionCheckpoint.model_validate_json(
            read_private_file(_checkpoint_path(data_dir, wallet_fingerprint))
        )
    except (OSError, ValidationError, ValueError):
        return unique_addresses
    if (
        checkpoint.version != 1
        or checkpoint.wallet_fingerprint != wallet_fingerprint
        or checkpoint.network != network
        or checkpoint.backend_id != _backend_id(backend)
    ):
        return unique_addresses
    return [
        address
        for address in unique_addresses
        if _address_hash(address) not in checkpoint.scanned_address_hashes
    ]

has_reconstruction_address_coverage(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, mixdepth_count: int, range_end: int) -> bool

Return whether a completed baseline covered every initial regular branch.

Source code in jmwallet/src/jmwallet/history_state.py 
 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
def has_reconstruction_address_coverage(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    mixdepth_count: int,
    range_end: int,
) -> bool:
    """Return whether a completed baseline covered every initial regular branch."""
    try:
        path = _checkpoint_path(data_dir, wallet_fingerprint)
        checkpoint = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
    except (OSError, ValidationError, ValueError):
        return False
    if (
        checkpoint.version != 1
        or checkpoint.wallet_fingerprint != wallet_fingerprint
        or checkpoint.network != network
        or checkpoint.backend_id != _backend_id(backend)
    ):
        return False
    return all(
        checkpoint.regular_branch_ends.get(f"{mixdepth}:{change}", -1) >= range_end
        for mixdepth in range(mixdepth_count)
        for change in (0, 1)
    )

has_reconstruction_branch_coverage(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, mixdepth: int, change: int, range_end: int) -> bool

Return whether one regular branch was historically scanned through an index.

Source code in jmwallet/src/jmwallet/history_state.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def has_reconstruction_branch_coverage(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    mixdepth: int,
    change: int,
    range_end: int,
) -> bool:
    """Return whether one regular branch was historically scanned through an index."""
    try:
        path = _checkpoint_path(data_dir, wallet_fingerprint)
        checkpoint = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
    except (OSError, ValidationError, ValueError):
        return False
    return (
        checkpoint.version == 1
        and checkpoint.wallet_fingerprint == wallet_fingerprint
        and checkpoint.network == network
        and checkpoint.backend_id == _backend_id(backend)
        and checkpoint.regular_branch_ends.get(f"{mixdepth}:{change}", -1) >= range_end
    )

load_reconstruction_cursor(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend) -> str | None

Load a matching completed baseline, falling back safely on any mismatch.

Source code in jmwallet/src/jmwallet/history_state.py 
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
def load_reconstruction_cursor(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
) -> str | None:
    """Load a matching completed baseline, falling back safely on any mismatch."""
    try:
        path = _checkpoint_path(data_dir, wallet_fingerprint)
    except ValueError as exc:
        logger.warning(f"Ignoring history reconstruction checkpoint: {exc}")
        return None
    if not path.exists():
        return None

    try:
        checkpoint = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
    except (OSError, ValidationError, ValueError) as exc:
        logger.warning("Ignoring invalid history reconstruction checkpoint")
        logger.bind(sensitive=True).warning(
            f"Ignoring invalid history reconstruction checkpoint {path}: {exc}"
        )
        return None

    if (
        checkpoint.version != 1
        or checkpoint.wallet_fingerprint != wallet_fingerprint
        or checkpoint.network != network
        or checkpoint.backend_id != _backend_id(backend)
    ):
        logger.debug("Ignoring mismatched history reconstruction checkpoint")
        logger.bind(sensitive=True).debug(
            f"Ignoring mismatched history reconstruction checkpoint {path}"
        )
        return None
    return checkpoint.cursor

record_reconstruction_address_coverage(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, branch_ends: Mapping[str, int]) -> None

Persist regular branches that completed historical scanning.

Source code in jmwallet/src/jmwallet/history_state.py 
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
def record_reconstruction_address_coverage(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    branch_ends: Mapping[str, int],
) -> None:
    """Persist regular branches that completed historical scanning."""
    path = _checkpoint_path(data_dir, wallet_fingerprint)
    cursor: str | None = None
    regular_branch_ends: dict[str, int] = {}
    scanned_address_hashes: set[str] = set()
    try:
        existing = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
        if (
            existing.wallet_fingerprint == wallet_fingerprint
            and existing.network == network
            and existing.backend_id == _backend_id(backend)
        ):
            cursor = existing.cursor
            regular_branch_ends.update(existing.regular_branch_ends)
            scanned_address_hashes.update(existing.scanned_address_hashes)
    except (OSError, ValidationError, ValueError):
        pass
    for key, index in branch_ends.items():
        regular_branch_ends[key] = max(regular_branch_ends.get(key, -1), index)
    checkpoint = ReconstructionCheckpoint(
        wallet_fingerprint=wallet_fingerprint,
        network=network,
        backend_id=_backend_id(backend),
        cursor=cursor,
        regular_branch_ends=regular_branch_ends,
        scanned_address_hashes=scanned_address_hashes,
    )
    atomic_write_private(path, checkpoint.model_dump_json(indent=2).encode("utf-8") + b"\n")

record_reconstruction_explicit_address_coverage(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, addresses: Iterable[str]) -> None

Persist explicit addresses that completed historical scanning.

Source code in jmwallet/src/jmwallet/history_state.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
def record_reconstruction_explicit_address_coverage(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    addresses: Iterable[str],
) -> None:
    """Persist explicit addresses that completed historical scanning."""
    path = _checkpoint_path(data_dir, wallet_fingerprint)
    cursor: str | None = None
    regular_branch_ends: dict[str, int] = {}
    scanned_address_hashes: set[str] = set()
    try:
        existing = ReconstructionCheckpoint.model_validate_json(read_private_file(path))
        if (
            existing.wallet_fingerprint == wallet_fingerprint
            and existing.network == network
            and existing.backend_id == _backend_id(backend)
        ):
            cursor = existing.cursor
            regular_branch_ends.update(existing.regular_branch_ends)
            scanned_address_hashes.update(existing.scanned_address_hashes)
    except (OSError, ValidationError, ValueError):
        pass
    scanned_address_hashes.update(_address_hash(address) for address in addresses)
    checkpoint = ReconstructionCheckpoint(
        wallet_fingerprint=wallet_fingerprint,
        network=network,
        backend_id=_backend_id(backend),
        cursor=cursor,
        regular_branch_ends=regular_branch_ends,
        scanned_address_hashes=scanned_address_hashes,
    )
    atomic_write_private(path, checkpoint.model_dump_json(indent=2).encode("utf-8") + b"\n")

save_reconstruction_cursor(data_dir: Path, *, wallet_fingerprint: str, network: str, backend: BlockchainBackend, cursor: str, address_paths: Mapping[str, tuple[int, int, int]] | None = None) -> None

Atomically persist a cursor after all corresponding rows are durable.

Source code in jmwallet/src/jmwallet/history_state.py 
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
def save_reconstruction_cursor(
    data_dir: Path,
    *,
    wallet_fingerprint: str,
    network: str,
    backend: BlockchainBackend,
    cursor: str,
    address_paths: Mapping[str, tuple[int, int, int]] | None = None,
) -> None:
    """Atomically persist a cursor after all corresponding rows are durable."""
    if not cursor:
        return
    regular_branch_ends: dict[str, int] = {}
    for mixdepth, change, index in (address_paths or {}).values():
        if change not in (0, 1):
            continue
        key = f"{mixdepth}:{change}"
        regular_branch_ends[key] = max(regular_branch_ends.get(key, -1), index)
    existing_coverage: dict[str, int] = {}
    existing_address_hashes: set[str] = set()
    try:
        existing = ReconstructionCheckpoint.model_validate_json(
            read_private_file(_checkpoint_path(data_dir, wallet_fingerprint))
        )
        if (
            existing.wallet_fingerprint == wallet_fingerprint
            and existing.network == network
            and existing.backend_id == _backend_id(backend)
        ):
            existing_coverage = existing.regular_branch_ends
            existing_address_hashes = existing.scanned_address_hashes
    except (OSError, ValidationError, ValueError):
        pass
    for key, index in existing_coverage.items():
        regular_branch_ends[key] = max(regular_branch_ends.get(key, -1), index)
    checkpoint = ReconstructionCheckpoint(
        wallet_fingerprint=wallet_fingerprint,
        network=network,
        backend_id=_backend_id(backend),
        cursor=cursor,
        regular_branch_ends=regular_branch_ends,
        scanned_address_hashes=existing_address_hashes,
    )
    path = _checkpoint_path(data_dir, wallet_fingerprint)
    atomic_write_private(path, checkpoint.model_dump_json(indent=2).encode("utf-8") + b"\n")