Coverage for bzfs_main/replication.py: 99%
908 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 18:44 +0000
1# Copyright 2024 Wolfgang Hoschek AT mac DOT com
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14#
15"""The core replication algorithm is in replicate_dataset(), which performs reliable full and/or incremental 'zfs send' and
16'zfs receive' operations on snapshots, using resumable ZFS sends when possible.
18For replication of multiple datasets, including recursive replication, see bzfs.py:replicate_datasets().
19"""
21from __future__ import (
22 annotations,
23)
24import base64
25import logging
26import os
27import re
28import shlex
29import subprocess
30import sys
31import threading
32import time
33from collections.abc import (
34 Iterable,
35 Iterator,
36 Mapping,
37)
38from concurrent.futures import (
39 Executor,
40 Future,
41)
42from subprocess import (
43 DEVNULL,
44 PIPE,
45)
46from typing import (
47 TYPE_CHECKING,
48 Final,
49 final,
50)
52from bzfs_main.argparse_actions import (
53 has_timerange_filter,
54)
55from bzfs_main.configuration import (
56 is_same_remote,
57)
58from bzfs_main.detect import (
59 POOL_GUID,
60 ZFS_VERSION_IS_AT_LEAST_2_1_0,
61 ZFS_VERSION_IS_AT_LEAST_2_2_0,
62 are_bookmarks_enabled,
63 is_zpool_feature_enabled_or_active,
64)
65from bzfs_main.filter import (
66 filter_properties,
67 filter_snapshots,
68)
69from bzfs_main.incremental_send_steps import (
70 incremental_send_steps,
71)
72from bzfs_main.parallel_batch_cmd import (
73 run_ssh_cmd_batched,
74 run_ssh_cmd_parallel,
75)
76from bzfs_main.progress_reporter import (
77 PV_FILE_THREAD_SEPARATOR,
78)
79from bzfs_main.util.connection import (
80 DEDICATED,
81 SHARED,
82 ConnectionPool,
83 dquote,
84 squote,
85 timeout,
86)
87from bzfs_main.util.parallel_iterator import (
88 parallel_iterator,
89 run_in_parallel,
90)
91from bzfs_main.util.retry import (
92 Retry,
93 RetryableError,
94)
95from bzfs_main.util.utils import (
96 DONT_SKIP_DATASET,
97 FILE_PERMISSIONS,
98 LOG_DEBUG,
99 LOG_TRACE,
100 Subprocesses,
101 append_if_absent,
102 cut,
103 die,
104 getenv_bool,
105 human_readable_bytes,
106 is_descendant,
107 list_formatter,
108 open_nofollow,
109 replace_prefix,
110 sha256_urlsafe_base64,
111 sort_datasets,
112 stderr_to_str,
113 xprint,
114)
116if TYPE_CHECKING: # pragma: no cover - for type hints only
117 from bzfs_main.bzfs import (
118 Job,
119 )
120 from bzfs_main.configuration import (
121 Params,
122 Remote,
123 )
126# constants:
127INJECT_DST_PIPE_FAIL_KBYTES: Final[int] = 400 # for testing only
128_RIGHT_JUST: Final[int] = 7
131def replicate_dataset(job: Job, src_dataset: str, tid: str, retry: Retry, *, has_descendant_dataset: bool) -> bool:
132 """Replicates src_dataset to dst_dataset (thread-safe); For replication of multiple datasets, including recursive
133 replication, see bzfs.py:replicate_datasets(); Return False to skip descendants; True to recurse into them."""
134 p, log = job.params, job.params.log
135 src, dst = p.src, p.dst
136 dst_dataset: str = replace_prefix(src_dataset, old_prefix=src.root_dataset, new_prefix=dst.root_dataset)
137 log.debug(p.dry(f"{tid} Replicating: %s"), f"{src_dataset} --> {dst_dataset} ...")
139 list_result: bool | tuple[list[str], list[str], list[str], dict[str, str], str, str, _Continuity | None] = (
140 _list_and_filter_src_and_dst_snapshots(job, src_dataset, dst_dataset, tid, has_descendant_dataset)
141 )
142 if isinstance(list_result, bool):
143 return list_result
144 (
145 basis_src_snapshots_with_guids,
146 _src_snapshots_with_guids,
147 dst_snapshots_with_guids,
148 included_src_guids,
149 latest_src_snapshot,
150 oldest_src_snapshot,
151 continuity,
152 ) = list_result
153 assert latest_src_snapshot
154 assert oldest_src_snapshot
155 latest_dst_snapshot: str = ""
156 latest_common_src_snapshot: str = ""
157 done_checking: bool = False
159 if job.dst_dataset_exists[dst_dataset]:
160 rollback_result: tuple[str, str, bool] = _rollback_dst_dataset_if_necessary(
161 job,
162 dst_dataset,
163 latest_src_snapshot,
164 basis_src_snapshots_with_guids,
165 dst_snapshots_with_guids,
166 done_checking,
167 tid,
168 )
169 latest_dst_snapshot, latest_common_src_snapshot, done_checking = rollback_result
170 if latest_src_snapshot and latest_src_snapshot == latest_common_src_snapshot:
171 log.info(f"{tid} Already up-to-date: %s", dst_dataset)
172 return True
174 log.debug("latest_common_src_snapshot: %s", latest_common_src_snapshot) # is a snapshot or bookmark
175 log.log(LOG_TRACE, "latest_dst_snapshot: %s", latest_dst_snapshot)
176 props_cache: dict[tuple[str, ...], dict[str, str | None]] = {} # fresh empty ZFS props cache for each dataset
177 dry_run_no_send: bool = False
178 if not latest_common_src_snapshot:
179 # no common snapshot exists; delete all dst snapshots and perform a full send of the oldest selected src snapshot
180 full_result: tuple[str, bool, bool] = _replicate_dataset_fully(
181 job,
182 src_dataset,
183 dst_dataset,
184 oldest_src_snapshot,
185 latest_src_snapshot,
186 latest_dst_snapshot,
187 included_src_guids,
188 dst_snapshots_with_guids,
189 continuity,
190 props_cache,
191 dry_run_no_send,
192 done_checking,
193 tid,
194 )
195 # we have now created a common snapshot
196 latest_common_src_snapshot, dry_run_no_send, done_checking = full_result
197 if latest_common_src_snapshot: 197 ↛ 236line 197 didn't jump to line 236 because the condition on line 197 was always true
198 # finally, incrementally replicate all selected snapshots from latest common snapshot until latest src snapshot
199 recv_resume_token_result: tuple[str | None, list[str], list[str]] = _recv_resume_token(job, dst_dataset)
200 recv_resume_token, _send_resume_opts, _recv_resume_opts = recv_resume_token_result
201 if recv_resume_token:
202 latest_common_src_snapshot = _replicate_dataset_incrementally(
203 job,
204 src_dataset,
205 dst_dataset,
206 latest_common_src_snapshot,
207 latest_src_snapshot,
208 basis_src_snapshots_with_guids,
209 included_src_guids,
210 recv_resume_token_result,
211 continuity,
212 props_cache,
213 dry_run_no_send,
214 done_checking,
215 tid,
216 )
217 recv_resume_token = None # we have now caught up
218 recv_resume_token_result = (recv_resume_token,) + recv_resume_token_result[1:]
219 dry_run_no_send = dry_run_no_send or p.dry_run
221 latest_common_src_snapshot = _replicate_dataset_incrementally(
222 job,
223 src_dataset,
224 dst_dataset,
225 latest_common_src_snapshot,
226 latest_src_snapshot,
227 basis_src_snapshots_with_guids,
228 included_src_guids,
229 recv_resume_token_result,
230 continuity,
231 props_cache,
232 dry_run_no_send,
233 done_checking,
234 tid,
235 )
236 return True
239def _list_and_filter_src_and_dst_snapshots(
240 job: Job, src_dataset: str, dst_dataset: str, tid: str, has_descendant: bool
241) -> bool | tuple[list[str], list[str], list[str], dict[str, str], str, str, _Continuity | None]:
242 """On replication, list and filter src and dst snapshots; return early with False to skip descendants; True to recurse
243 into them."""
244 p, log = job.params, job.params.log
245 src, dst = p.src, p.dst
247 # list GUID and name for dst snapshots, sorted ascending by createtxg (more precise than creation time)
248 dst_cmd: list[str] = p.split_args(f"{p.zfs_program} list -t snapshot -d 1 -s createtxg -Hp -o guid,name", dst_dataset)
250 # list GUID and name for src snapshots + bookmarks, primarily sort ascending by transaction group (which is more
251 # precise than creation time), secondarily sort such that snapshots appear after bookmarks for the same GUID.
252 # Note: A snapshot and its ZFS bookmarks always have the same GUID, creation time and transaction group. A snapshot
253 # changes its transaction group but retains its creation time and GUID on 'zfs receive' on another pool, i.e.
254 # comparing createtxg is only meaningful within a single pool, not across pools from src to dst. Comparing creation
255 # time remains meaningful across pools from src to dst. Creation time is a UTC Unix time in integer seconds.
256 # Note that 'zfs create', 'zfs snapshot' and 'zfs bookmark' CLIs enforce that snapshot names must not contain a '#'
257 # char, bookmark names must not contain a '@' char, and dataset names must not contain a '#' or '@' char.
258 # GUID and creation time also do not contain a '#' or '@' char.
259 filter_needs_creation_time: bool = has_timerange_filter(p.snapshot_filters)
260 types: str = "snapshot"
261 types += ",bookmark" if (p.use_bookmark or p.create_bookmarks != "none") and are_bookmarks_enabled(p, src) else ""
262 props: str = job.creation_prefix + "creation,guid,name" if filter_needs_creation_time else "guid,name"
263 src_cmd = p.split_args(f"{p.zfs_program} list -t {types} -s createtxg -s type -d 1 -Hp -o {props}", src_dataset)
264 job.maybe_inject_delete(src, dataset=src_dataset, delete_trigger="zfs_list_snapshot_src")
265 src_snapshots_and_bookmarks, dst_snapshots_with_guids_str = run_in_parallel( # list src+dst snapshots in parallel
266 lambda: job.try_ssh_command(src, LOG_TRACE, cmd=src_cmd),
267 lambda: job.try_ssh_command(dst, LOG_TRACE, cmd=dst_cmd, error_trigger="zfs_list_snapshot_dst"),
268 )
269 job.dst_dataset_exists[dst_dataset] = dst_snapshots_with_guids_str is not None
270 dst_snapshots_with_guids: list[str] = (dst_snapshots_with_guids_str or "").splitlines()
271 if src_snapshots_and_bookmarks is None:
272 log.warning("Third party deleted source: %s", src_dataset)
273 return False # src dataset has been deleted by some third party while we're running - nothing to do anymore
274 raw_src_snapshots_with_guids: list[str] = src_snapshots_and_bookmarks.splitlines()
275 src_snapshots_with_guids: list[str] = raw_src_snapshots_with_guids.copy()
276 if not p.use_bookmark:
277 src_snapshots_with_guids = [snapshot for snapshot in src_snapshots_with_guids if "@" in snapshot]
278 src_snapshots_and_bookmarks = None
279 if len(dst_snapshots_with_guids) == 0 and "bookmark" in types:
280 # src bookmarks serve no purpose if the destination dataset has no snapshot; ignore them
281 src_snapshots_with_guids = [snapshot for snapshot in src_snapshots_with_guids if "@" in snapshot]
282 num_src_snapshots_found: int = sum(1 for snapshot in src_snapshots_with_guids if "@" in snapshot)
283 with job.stats_lock:
284 job.num_snapshots_found += num_src_snapshots_found
285 # apply include/exclude regexes to ignore irrelevant src snapshots
286 basis_src_snapshots_with_guids: list[str] = src_snapshots_with_guids
287 src_snapshots_with_guids = filter_snapshots(job, src_snapshots_with_guids)
288 if filter_needs_creation_time:
289 raw_src_snapshots_with_guids = cut(field=2, lines=raw_src_snapshots_with_guids)
290 src_snapshots_with_guids = cut(field=2, lines=src_snapshots_with_guids)
291 basis_src_snapshots_with_guids = cut(field=2, lines=basis_src_snapshots_with_guids)
293 # find oldest and latest selected "true" snapshot, and map selected snapshot GUIDs to names.
294 # a snapshot is "true" if it is not a bookmark.
295 oldest_src_snapshot: str = ""
296 latest_src_snapshot: str = ""
297 included_src_guids: dict[str, str] = {}
298 for line in src_snapshots_with_guids:
299 guid, snapshot = line.split("\t", 1)
300 assert guid
301 assert snapshot
302 if "@" in snapshot:
303 included_src_guids[guid] = snapshot
304 latest_src_snapshot = snapshot
305 if not oldest_src_snapshot:
306 oldest_src_snapshot = snapshot
307 if len(src_snapshots_with_guids) == 0:
308 if p.skip_missing_snapshots == "fail":
309 die(f"Source dataset includes no snapshot: {src_dataset!r}. Consider using --skip-missing-snapshots=dataset")
310 elif p.skip_missing_snapshots == "dataset":
311 log.warning("Skipping source dataset because it includes no snapshot: %s", src_dataset)
312 recurse = has_descendant and job.dst_dataset_exists[dst_dataset] and not is_inconsistent(job, dst, dst_dataset)
313 if has_descendant and not recurse:
314 log.warning("Skipping descendant datasets as dst dataset does not exist or is inconsistent: %s", dst_dataset)
315 return recurse
316 log.debug("latest_src_snapshot: %s", latest_src_snapshot)
317 if latest_src_snapshot == "":
318 recurse = has_descendant and not (job.dst_dataset_exists[dst_dataset] and is_inconsistent(job, dst, dst_dataset))
319 if has_descendant and not recurse:
320 die(
321 f"Cannot replicate descendant datasets as destination dataset is inconsistent: {dst_dataset!r} "
322 f"and source dataset selects no snapshot: {src_dataset!r}. Resolve the interrupted receive before retrying."
323 )
324 log.info(f"{tid} Already-up-to-date: %s", dst_dataset)
325 return recurse
326 if p.create_bookmarks != "none" and are_bookmarks_enabled(p, src) and not p.dry_run:
327 continuity: _Continuity | None = _Continuity(
328 p,
329 src_dataset,
330 dst_dataset,
331 raw_src_snapshots_with_guids,
332 dst_snapshots_with_guids,
333 )
334 else:
335 continuity = None
336 return (
337 basis_src_snapshots_with_guids,
338 src_snapshots_with_guids,
339 dst_snapshots_with_guids,
340 included_src_guids,
341 latest_src_snapshot,
342 oldest_src_snapshot,
343 continuity,
344 )
347def _rollback_dst_dataset_if_necessary(
348 job: Job,
349 dst_dataset: str,
350 latest_src_snapshot: str,
351 src_snapshots_with_guids: list[str],
352 dst_snapshots_with_guids: list[str],
353 done_checking: bool,
354 tid: str,
355) -> tuple[str, str, bool]:
356 """On replication, rollback dst if necessary; error out if not permitted."""
357 p, log = job.params, job.params.log
358 dst = p.dst
359 latest_dst_snapshot: str = ""
360 latest_dst_guid: str = ""
361 if len(dst_snapshots_with_guids) > 0:
362 latest_dst_guid, latest_dst_snapshot = dst_snapshots_with_guids[-1].split("\t", 1)
363 if p.force_rollback_to_latest_snapshot:
364 log.info(p.dry(f"{tid} Rolling back destination to most recent snapshot: %s"), latest_dst_snapshot)
365 # rollback just in case the dst dataset was modified since its most recent snapshot
366 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset)
367 cmd: list[str] = p.split_args(f"{dst.sudo} {p.zfs_program} rollback", latest_dst_snapshot)
368 job.try_ssh_command(dst, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd, exists=False)
370 # find most recent snapshot (or bookmark) that src and dst have in common - we'll start to replicate
371 # from there up to the most recent src snapshot. any two snapshots are "common" iff their ZFS GUIDs (i.e.
372 # contents) are equal. See https://github.com/openzfs/zfs/commit/305bc4b370b20de81eaf10a1cf724374258b74d1
373 def latest_common_snapshot(snapshots_with_guids: list[str], intersect_guids: set[str]) -> tuple[str | None, str]:
374 """Returns a true snapshot instead of its bookmark with the same GUID, per the sort order previously used for 'zfs
375 list -s ...'."""
376 for _line in reversed(snapshots_with_guids):
377 guid_, snapshot_ = _line.split("\t", 1)
378 if guid_ in intersect_guids:
379 return guid_, snapshot_ # can be a snapshot or bookmark
380 return None, ""
382 latest_common_guid, latest_common_src_snapshot = latest_common_snapshot(
383 src_snapshots_with_guids, set(cut(field=1, lines=dst_snapshots_with_guids))
384 )
385 log.debug("latest_common_src_snapshot: %s", latest_common_src_snapshot) # is a snapshot or bookmark
386 log.log(LOG_TRACE, "latest_dst_snapshot: %s", latest_dst_snapshot)
388 if latest_common_src_snapshot and latest_common_guid != latest_dst_guid:
389 # found latest common snapshot but dst has an even newer snapshot. rollback dst to that common snapshot.
390 assert latest_common_guid
391 _, latest_common_dst_snapshot = latest_common_snapshot(dst_snapshots_with_guids, {latest_common_guid})
392 assert latest_common_dst_snapshot
393 if not (p.force_rollback_to_latest_common_snapshot or p.force):
394 die(
395 f"Conflict: Most recent destination snapshot {latest_dst_snapshot} is more recent than "
396 f"most recent common snapshot {latest_common_dst_snapshot}. Rollback destination first, "
397 "for example via --force-rollback-to-latest-common-snapshot (or --force) option."
398 )
399 if p.force_once:
400 p.force.value = False
401 p.force_rollback_to_latest_common_snapshot.value = False
402 log.info(p.dry(f"{tid} Rolling back destination to most recent common snapshot: %s"), latest_common_dst_snapshot)
403 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset)
404 cmd = p.split_args(
405 f"{dst.sudo} {p.zfs_program} rollback -r {p.force_unmount} {p.force_hard}", latest_common_dst_snapshot
406 )
407 try:
408 job.run_ssh_command(dst, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd)
409 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
410 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
411 retry_immediately_once: bool = _clear_resumable_recv_state_if_necessary(job, dst_dataset, stderr)
412 # op isn't idempotent so retries regather current state from the start of replicate_dataset()
413 raise RetryableError(display_msg="zfs rollback", retry_immediately_once=retry_immediately_once) from e
415 return latest_dst_snapshot, latest_common_src_snapshot, done_checking
418def _replicate_dataset_fully(
419 job: Job,
420 src_dataset: str,
421 dst_dataset: str,
422 oldest_src_snapshot: str,
423 latest_src_snapshot: str,
424 latest_dst_snapshot: str,
425 included_src_guids: Mapping[str, str],
426 dst_snapshots_with_guids: list[str],
427 continuity: _Continuity | None,
428 props_cache: dict[tuple[str, ...], dict[str, str | None]],
429 dry_run_no_send: bool,
430 done_checking: bool,
431 tid: str,
432) -> tuple[str, bool, bool]:
433 """On replication, deletes all dst snapshots and performs a full send of the oldest selected src snapshot, which in turn
434 creates a common snapshot; error out if not permitted."""
435 p, log = job.params, job.params.log
436 src, dst = p.src, p.dst
437 latest_common_src_snapshot: str = ""
438 if latest_dst_snapshot:
439 if not p.force:
440 die(
441 f"Conflict: No common snapshot found between {src_dataset} and {dst_dataset} even though "
442 "destination has at least one snapshot. Aborting. Consider using --force option to first "
443 "delete all existing destination snapshots in order to be able to proceed with replication."
444 )
445 if p.force_once: 445 ↛ 448line 445 didn't jump to line 448 because the condition on line 445 was always true
446 p.force.value = False
447 p.force_rollback_to_latest_common_snapshot.value = False
448 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset)
449 # extract SNAPSHOT_TAG from GUID<TAB>DATASET@SNAPSHOT_TAG
450 delete_snapshots(job, dst, dst_dataset, snapshot_tags=cut(2, separator="@", lines=dst_snapshots_with_guids))
451 if p.dry_run:
452 # As we're in --dryrun (--force) mode this conflict resolution step (see above) wasn't really executed:
453 # "no common snapshot was found. delete all dst snapshots". In turn, this would cause the subsequent
454 # 'zfs receive -n' to fail with "cannot receive new filesystem stream: destination has snapshots; must
455 # destroy them to overwrite it". So we skip the zfs send/receive step and keep on trucking.
456 dry_run_no_send = True
458 # to start with, fully replicate oldest snapshot, which in turn creates a common snapshot
459 if p.no_stream:
460 oldest_src_snapshot = latest_src_snapshot
461 if oldest_src_snapshot: 461 ↛ 507line 461 didn't jump to line 507 because the condition on line 461 was always true
462 if not job.dst_dataset_exists[dst_dataset]:
463 # on destination, create parent filesystem and ancestors if they do not yet exist
464 dst_dataset_parent: str = os.path.dirname(dst_dataset)
465 if not job.dst_dataset_exists[dst_dataset_parent]:
466 if p.dry_run:
467 dry_run_no_send = True
468 if dst_dataset_parent: 468 ↛ 471line 468 didn't jump to line 471 because the condition on line 468 was always true
469 _create_zfs_filesystem(job, dst_dataset_parent)
471 recv_resume_token_result: tuple[str | None, list[str], list[str]] = _recv_resume_token(job, dst_dataset)
472 recv_resume_token, send_resume_opts, recv_resume_opts = recv_resume_token_result
473 curr_size: int = _estimate_send_size(job, src, dst_dataset, recv_resume_token, oldest_src_snapshot)
474 humansize: str = _format_size(curr_size)
475 if recv_resume_token:
476 oldest_src_snapshot = _decode_resume_token(job, recv_resume_token, src_dataset, dst_dataset, included_src_guids)
477 send_opts: list[str] = p.curr_zfs_send_resume_opts + send_resume_opts # e.g. curr + ["-t", "1-c740b4779-..."]
478 else:
479 send_opts = p.curr_zfs_send_program_opts + [oldest_src_snapshot]
480 send_cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} send", send_opts)
481 recv_opts: list[str] = p.zfs_full_recv_opts.copy() + recv_resume_opts
482 recv_opts, set_opts = _add_recv_property_options(job, True, recv_opts, src_dataset, props_cache)
483 recv_cmd: list[str] = p.split_args(
484 f"{dst.sudo} {p.zfs_program} receive -F", p.dry_run_recv, recv_opts, dst_dataset, allow_all=True
485 )
486 log.info(p.dry(f"{tid} Full send: %s"), f"{oldest_src_snapshot} --> {dst_dataset} ({humansize.strip()}) ...")
487 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset)
488 dry_run_no_send = dry_run_no_send or p.dry_run_no_send
489 job.maybe_inject_params(error_trigger="full_zfs_send_params")
490 humansize = humansize.rjust(_RIGHT_JUST * 3 + 2)
491 if continuity is not None:
492 continuity.create_tmp_bookmarks(job, src, [oldest_src_snapshot])
493 _run_zfs_send_receive( # do the real work
494 job, src_dataset, dst_dataset, send_cmd, recv_cmd, curr_size, humansize, dry_run_no_send, "full_zfs_send"
495 )
496 latest_common_src_snapshot = oldest_src_snapshot # we have now created a common snapshot
497 if not p.dry_run:
498 job.dst_dataset_exists[dst_dataset] = True
499 with job.stats_lock:
500 job.num_snapshots_replicated += 1
501 if continuity is not None:
502 continuity.mark_snapshots_as_replicated([oldest_src_snapshot])
503 continuity.promote_and_gc_bookmarks(job, src)
504 _zfs_set(job, set_opts, dst, dst_dataset)
505 dry_run_no_send = dry_run_no_send or p.dry_run
507 return latest_common_src_snapshot, dry_run_no_send, done_checking
510def _replicate_dataset_incrementally(
511 job: Job,
512 src_dataset: str,
513 dst_dataset: str,
514 latest_common_src_snapshot: str,
515 latest_src_snapshot: str,
516 basis_src_snapshots_with_guids: list[str],
517 included_src_guids: Mapping[str, str],
518 recv_resume_token_result: tuple[str | None, list[str], list[str]],
519 continuity: _Continuity | None,
520 props_cache: dict[tuple[str, ...], dict[str, str | None]],
521 dry_run_no_send: bool,
522 done_checking: bool,
523 tid: str,
524) -> str:
525 """Incrementally replicates all selected snapshots from latest common snapshot until latest src snapshot."""
526 p, log = job.params, job.params.log
527 src, dst = p.src, p.dst
528 recv_resume_token, send_resume_opts, recv_resume_opts = recv_resume_token_result
529 set_opts: list[str] = []
531 def replication_candidates() -> tuple[list[str], list[str]]:
532 assert len(basis_src_snapshots_with_guids) > 0
533 result_snapshots: list[str] = []
534 result_guids: list[str] = []
535 last_appended_guid: str = ""
536 snapshot_itr: Iterator[str] = reversed(basis_src_snapshots_with_guids)
537 while True:
538 guid, snapshot = next(snapshot_itr).split("\t", 1)
539 if "@" in snapshot:
540 result_snapshots.append(snapshot)
541 result_guids.append(guid)
542 last_appended_guid = guid
543 if snapshot == latest_common_src_snapshot: # latest_common_src_snapshot is a snapshot or bookmark
544 if guid != last_appended_guid and "@" not in snapshot:
545 # only appends the src bookmark if it has no snapshot. If the bookmark has a snapshot then that
546 # snapshot has already been appended, per the sort order previously used for 'zfs list -s ...'
547 result_snapshots.append(snapshot)
548 result_guids.append(guid)
549 break
550 result_snapshots.reverse()
551 result_guids.reverse()
552 assert len(result_snapshots) > 0
553 assert len(result_snapshots) == len(result_guids)
554 return result_guids, result_snapshots
556 if recv_resume_token:
557 estimate_send_sizes: list[int] = [_estimate_send_size(job, src, dst_dataset, recv_resume_token)]
558 decoded_src_snapshot: str = _decode_resume_token(
559 job, recv_resume_token, src_dataset, dst_dataset, included_src_guids
560 )
561 steps_todo: list[tuple[str, str, str, list[str]]] = [
562 ("-i", latest_common_src_snapshot, decoded_src_snapshot, [decoded_src_snapshot])
563 ]
564 else:
565 # collect the most recent common snapshot (which may be a bookmark) followed by all src snapshots
566 # (that are not a bookmark) that are more recent than that.
567 cand_guids, cand_snapshots = replication_candidates()
568 if len(cand_snapshots) == 1:
569 # latest_src_snapshot is a (true) snapshot that is equal to latest_common_src_snapshot or LESS recent
570 # than latest_common_src_snapshot. The latter case can happen if latest_common_src_snapshot is a
571 # bookmark whose snapshot has been deleted on src.
572 return latest_common_src_snapshot # nothing more tbd
573 if p.no_stream:
574 if not any(guid in included_src_guids for guid in cand_guids[1:]):
575 # The cand_snapshots list (and basis_src_snapshots_with_guids) can contain snapshots that are newer than
576 # what the user selected via --include/exclude-snapshot-*. These newer snapshots are irrelevant here.
577 # Example: A resume token represents @s3, and @s4 exists on src but is excluded by --exclude-snapshot-*.
578 # Replanning from @s3 sees [@s3, @s4] in cand_snapshots, but no selected snapshot remains after @s3, so
579 # replication is done.
580 return latest_common_src_snapshot # nothing more tbd
581 # skip intermediate snapshots
582 steps_todo = [("-i", latest_common_src_snapshot, latest_src_snapshot, [latest_src_snapshot])]
583 else:
584 # include intermediate src snapshots that pass --{include,exclude}-snapshot-* policy, using
585 # a series of -i/-I send/receive steps that skip excluded src snapshots.
586 steps_todo = _incremental_send_steps_wrapper(p, cand_snapshots, cand_guids, set(included_src_guids), False)
587 estimate_send_sizes = _estimate_send_sizes_in_parallel(job, src, dst_dataset, recv_resume_token, steps_todo)
589 log.log(LOG_TRACE, "steps_todo: %s", list_formatter(steps_todo, "; "))
590 total_size: int = sum(estimate_send_sizes)
591 total_num: int = sum(len(to_snapshots) for incr_flag, from_snap, to_snap, to_snapshots in steps_todo)
592 done_size: int = 0
593 done_num: int = 0
594 for i, (incr_flag, from_snap, to_snap, to_snapshots) in enumerate(steps_todo):
595 curr_num_snapshots: int = len(to_snapshots)
596 curr_size: int = estimate_send_sizes[i]
597 humansize: str = _format_size(total_size) + "/" + _format_size(done_size) + "/" + _format_size(curr_size)
598 human_num: str = f"{total_num}/{done_num}/{curr_num_snapshots} snapshots"
599 if recv_resume_token:
600 send_opts: list[str] = p.curr_zfs_send_resume_opts + send_resume_opts # e.g. curr + ["-t", "1-c740b4779-..."]
601 else:
602 send_opts = p.curr_zfs_send_program_opts + [incr_flag, from_snap, to_snap]
603 send_cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} send", send_opts)
604 recv_opts: list[str] = p.zfs_recv_program_opts.copy() + recv_resume_opts
605 recv_opts, set_opts = _add_recv_property_options(job, False, recv_opts, src_dataset, props_cache)
606 recv_cmd: list[str] = p.split_args(
607 f"{dst.sudo} {p.zfs_program} receive", p.dry_run_recv, recv_opts, dst_dataset, allow_all=True
608 )
609 dense_size: str = p.two_or_more_spaces_regex.sub("", humansize.strip())
610 log.info(
611 p.dry(f"{tid} Incremental send {incr_flag}: %s"),
612 f"{from_snap} .. {to_snap[to_snap.index('@'):]} --> {dst_dataset} ({dense_size}) ({human_num}) ...",
613 )
614 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset, busy_if_send=False)
615 if p.dry_run and not job.dst_dataset_exists[dst_dataset]:
616 dry_run_no_send = True
617 dry_run_no_send = dry_run_no_send or p.dry_run_no_send
619 assert p.create_bookmarks
620 snaps_to_bookmark: list[str] = []
621 if p.create_bookmarks == "all":
622 snaps_to_bookmark = to_snapshots
623 elif p.create_bookmarks != "none":
624 threshold_millis: int = p.xperiods.label_milliseconds("_" + p.create_bookmarks)
625 snaps_to_bookmark = [snap for snap in to_snapshots if p.xperiods.label_milliseconds(snap) >= threshold_millis]
626 if i == len(steps_todo) - 1 and (len(snaps_to_bookmark) == 0 or snaps_to_bookmark[-1] != to_snap):
627 snaps_to_bookmark.append(to_snap) # ensure latest common snapshot is bookmarked
628 if continuity is not None:
629 continuity.create_tmp_bookmarks(job, src, snaps_to_bookmark)
631 job.maybe_inject_params(error_trigger="incr_zfs_send_params")
632 _run_zfs_send_receive( # do the real work
633 job, src_dataset, dst_dataset, send_cmd, recv_cmd, curr_size, humansize, dry_run_no_send, "incr_zfs_send"
634 )
635 done_size += curr_size
636 done_num += curr_num_snapshots
637 latest_common_src_snapshot = to_snap
638 with job.stats_lock:
639 job.num_snapshots_replicated += curr_num_snapshots
640 if continuity is not None:
641 continuity.mark_snapshots_as_replicated(to_snapshots)
642 if continuity is not None:
643 continuity.promote_and_gc_bookmarks(job, src)
644 _zfs_set(job, set_opts, dst, dst_dataset)
645 return latest_common_src_snapshot
648def _format_size(num_bytes: int) -> str:
649 """Formats a byte count for human-readable logs."""
650 return human_readable_bytes(num_bytes, separator="").rjust(_RIGHT_JUST)
653def _prepare_zfs_send_receive(
654 job: Job,
655 src_dataset: str,
656 send_cmd: list[str],
657 recv_cmd: list[str],
658 size_estimate_bytes: int,
659 size_estimate_human: str,
660 src_ssh_cmd: tuple[str, ...],
661 dst_ssh_cmd: tuple[str, ...],
662) -> str:
663 """Constructs zfs send/recv pipelines with optional compression, mbuffer and pv."""
664 p = job.params
665 src, dst = p.src, p.dst
666 send_cmd_str: str = shlex.join(send_cmd)
667 recv_cmd_str: str = shlex.join(recv_cmd)
668 recordsize: int = abs(job.src_properties[src_dataset].recordsize)
669 same_remote: bool = is_same_remote(src, dst)
670 r2r_mode: str = p.r2r_mode
671 assert r2r_mode in ("off", "pull", "push"), r2r_mode
673 is_large_enough: bool = p.no_estimate_send_size or size_estimate_bytes >= p.min_pipe_transfer_size
674 is_networked: bool = src.is_nonlocal or dst.is_nonlocal
675 src_has_shell: bool = p.is_program_available("sh", "src")
676 dst_has_shell: bool = p.is_program_available("sh", "dst")
677 use_src_pipeline: bool = src_has_shell and (r2r_mode == "off" or not same_remote)
678 use_dst_pipeline: bool = dst_has_shell and (r2r_mode == "off" or not same_remote)
679 use_compression: bool = p.is_program_available("zstd", "src") and p.is_program_available("zstd", "dst")
680 use_compression = use_compression and is_large_enough and is_networked and use_src_pipeline and use_dst_pipeline
681 use_src_buffer: bool = is_large_enough and is_networked and use_src_pipeline and p.is_program_available("mbuffer", "src")
682 use_dst_buffer: bool = is_large_enough and is_networked and use_dst_pipeline and p.is_program_available("mbuffer", "dst")
683 use_local_buffer: bool = p.is_program_available("mbuffer", "local")
684 use_local_buffer = use_local_buffer and r2r_mode == "off" and is_large_enough and src.is_nonlocal and dst.is_nonlocal
686 pv_location: str = ""
687 if r2r_mode == "off" and p.is_program_available("pv", "local"):
688 if not src.ssh_user_host:
689 pv_location = "src"
690 elif not dst.ssh_user_host:
691 pv_location = "dst"
692 else:
693 pv_location = "local"
695 # assemble pipeline running on source leg
696 src_stages: list[str] = []
697 src_pipe_fail_offset = job.inject_params.pop("inject_src_pipe_fail_offset", None)
698 if src_pipe_fail_offset is not None:
699 assert isinstance(src_pipe_fail_offset, int)
700 src_stages.append(f"(dd bs=1 count={src_pipe_fail_offset} 2>/dev/null && false)")
701 elif job.inject_params.pop(f"inject_src_pipe_fail_{src_dataset}", False):
702 # for testing; forward enough bytes that zfs receive can usually leave resumable state
703 src_stages.append(f"(dd bs=1024 count={INJECT_DST_PIPE_FAIL_KBYTES} 2>/dev/null && false)")
704 elif job.inject_params.get("inject_src_pipe_fail", False):
705 # for testing; initially forward some bytes and then fail
706 src_stages.append("(dd bs=64 count=1 2>/dev/null && false)")
707 if job.inject_params.get("inject_src_pipe_garble", False):
708 src_stages.append("gzip -1 -c -n") # for testing; forward garbled bytes
709 if pv_location == "src":
710 src_stages.append(_pv_cmd(job, size_estimate_bytes, size_estimate_human))
711 if use_compression:
712 src_stages.append(_compress_cmd(p))
713 if use_src_buffer:
714 src_stages.append(_mbuffer_cmd(p, "src", recordsize))
715 if job.inject_params.get("inject_src_send_error", False):
716 send_cmd_str = f"{send_cmd_str} --injectedGarbageParameter" # for testing; induce CLI parse error
717 src_pipe: str = " | ".join([send_cmd_str] + src_stages)
718 if len(src_stages) > 0 and src.ssh_user_host:
719 src_pipe = p.shell_program + " -c " + dquote(src_pipe)
721 # assemble pipeline running on destination leg
722 dst_stages: list[str] = []
723 if use_dst_buffer:
724 dst_stages.append(_mbuffer_cmd(p, "dst", recordsize))
725 if use_compression:
726 dst_stages.append(_decompress_cmd(p))
727 if pv_location == "dst":
728 dst_stages.append(_pv_cmd(job, size_estimate_bytes, size_estimate_human))
729 if job.inject_params.get("inject_dst_pipe_fail", False):
730 # interrupt zfs receive for testing retry/resume; initially forward some bytes and then stop forwarding
731 dst_stages.append(f"dd bs=1024 count={INJECT_DST_PIPE_FAIL_KBYTES} 2>/dev/null")
732 if job.inject_params.get("inject_dst_pipe_garble", False):
733 dst_stages.append("gzip -1 -c -n") # for testing; forward garbled bytes
734 if job.inject_params.get("inject_dst_receive_error", False):
735 recv_cmd_str = f"{recv_cmd_str} --injectedGarbageParameter" # for testing; induce CLI parse error
736 dst_pipe: str = " | ".join(dst_stages + [recv_cmd_str])
737 if len(dst_stages) > 0 and dst.ssh_user_host:
738 dst_pipe = p.shell_program + " -c " + dquote(dst_pipe)
740 if r2r_mode == "off":
741 # If there's no support for shell pipelines, we can't do compression, mbuffering, monitoring and rate-limiting,
742 # so we fall back to simple zfs send/receive.
743 if not src_has_shell:
744 src_pipe = send_cmd_str
745 if not dst_has_shell:
746 dst_pipe = recv_cmd_str
748 # assemble pipeline running on middle leg between source and destination. only enabled for pull-push mode
749 local_stages: list[str] = []
750 if use_local_buffer:
751 local_stages.append(_mbuffer_cmd(p, "local", recordsize))
752 if pv_location == "local":
753 # pull-push mode with compression enabled: reporting "percent complete" isn't straightforward because
754 # localhost observes the compressed data instead of the uncompressed data, so we disable the progress bar.
755 local_stages.append(_pv_cmd(job, size_estimate_bytes, size_estimate_human, disable_progress_bar=use_compression))
756 if use_local_buffer:
757 local_stages.append(_mbuffer_cmd(p, "local", recordsize))
759 src_leg: str = " ".join(src_ssh_cmd + (squote(src, src_pipe),))
760 dst_leg: str = " ".join(dst_ssh_cmd + (squote(dst, dst_pipe),))
761 return " | ".join([src_leg] + local_stages + [dst_leg])
763 def nested_ssh_cmd(remote: Remote) -> str:
764 """Builds nested ssh command text for r2r leg."""
765 if same_remote:
766 # Perf: use local replication if initiator sees src+dst have same user@host, port and ssh configfile.
767 # Example: ssh alice@src.example.com "sh -c 'zfs send <src_dataset> | zfs receive <dst_dataset>'"
768 return ""
769 cmd: list[str] = [p.ssh_program]
770 cmd += remote.ssh_extra_opts
771 if remote.ssh_config_file:
772 cmd += ["-F", remote.ssh_config_file]
773 if remote.ssh_cipher:
774 cmd += ["-c", remote.ssh_cipher]
775 if remote.ssh_port:
776 cmd += ["-p", str(remote.ssh_port)]
777 assert remote.ssh_user_host
778 cmd.append(remote.ssh_user_host)
779 return shlex.join(cmd)
781 if r2r_mode == "push":
782 # Example: ssh alice@src.example.com "sh -c 'zfs send ... | ssh bob@dst.example.com zfs receive ...'"
783 dst_pipe = dst_pipe if same_remote else shlex.quote(dst_pipe)
784 outer_ssh_cmd: tuple[str, ...] = src_ssh_cmd
785 r2r_cmd: str = f"{src_pipe} | {nested_ssh_cmd(dst)} {dst_pipe}"
786 else:
787 assert r2r_mode == "pull", r2r_mode
788 # Example: ssh bob@dst.example.com "sh -c 'ssh alice@src.example.com zfs send ... | zfs receive ...'"
789 src_pipe = src_pipe if same_remote else shlex.quote(src_pipe)
790 outer_ssh_cmd = dst_ssh_cmd
791 r2r_cmd = f"{nested_ssh_cmd(src)} {src_pipe} | {dst_pipe}"
792 r2r_cmd = shlex.quote(p.shell_program + " -c " + shlex.quote(r2r_cmd))
793 return " ".join(outer_ssh_cmd + (r2r_cmd,))
796def _run_zfs_send_receive(
797 job: Job,
798 src_dataset: str,
799 dst_dataset: str,
800 send_cmd: list[str],
801 recv_cmd: list[str],
802 size_estimate_bytes: int,
803 size_estimate_human: str,
804 dry_run_no_send: bool,
805 error_trigger: str | None = None,
806) -> None:
807 """Executes a zfs send/receive pipeline between source and destination."""
808 p, log = job.params, job.params.log
809 r2r_mode: str = p.r2r_mode
810 assert r2r_mode in ("off", "pull", "push"), r2r_mode
811 log.log(LOG_TRACE, "r2r_mode: %s", r2r_mode)
812 conn_pool_name: str = DEDICATED if p.dedicated_tcp_connection_per_zfs_send and r2r_mode == "off" else SHARED
813 src_conn_pool: ConnectionPool = p.connection_pools[p.src.location].pool(conn_pool_name)
814 dst_conn_pool: ConnectionPool = p.connection_pools[p.dst.location].pool(conn_pool_name)
815 with src_conn_pool.connection() as src_conn, dst_conn_pool.connection() as dst_conn:
816 src_ssh_cmd: tuple[str, ...] = src_conn.ssh_cmd_quoted
817 dst_ssh_cmd: tuple[str, ...] = dst_conn.ssh_cmd_quoted
818 cmd_str: str = _prepare_zfs_send_receive(
819 job, src_dataset, send_cmd, recv_cmd, size_estimate_bytes, size_estimate_human, src_ssh_cmd, dst_ssh_cmd
820 )
821 src_conn.refresh_ssh_connection_if_necessary(job)
822 dst_conn.refresh_ssh_connection_if_necessary(job)
823 cmd = [p.shell_program_local, "-c", cmd_str]
824 msg: str = "Would execute: %s" if dry_run_no_send else "Executing: %s"
825 log.debug(msg, cmd_str.lstrip())
826 if not dry_run_no_send:
827 try:
828 job.maybe_inject_error(cmd=cmd, error_trigger=error_trigger)
829 sp: Subprocesses = job.subprocesses
830 process = sp.subprocess_run(
831 cmd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True, timeout=timeout(job), check=True, log=log
832 )
833 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
834 retry_immediately_once: bool = False
835 if not isinstance(e, UnicodeDecodeError): 835 ↛ 838line 835 didn't jump to line 838 because the condition on line 835 was always true
836 xprint(log, stderr_to_str(e.stdout), file=sys.stdout)
837 log.warning("%s", stderr_to_str(e.stderr).rstrip())
838 if isinstance(e, subprocess.CalledProcessError): 838 ↛ 843line 838 didn't jump to line 843 because the condition on line 838 was always true
839 retry_immediately_once = _clear_resumable_recv_state_if_necessary(
840 job, dst_dataset, stderr_to_str(e.stderr)
841 )
842 # op isn't idempotent so retries regather current state from the start of replicate_dataset()
843 raise RetryableError(display_msg="zfs send/receive", retry_immediately_once=retry_immediately_once) from e
844 else:
845 xprint(log, process.stdout, file=sys.stdout)
846 xprint(log, process.stderr, file=sys.stderr)
847 if match := re.search(
848 r"(cannot send '[^'\r\n]+': |WARNING: could not send [^@\r\n]+@[^@\r\n]+:( |\r?\n))([^\r\n]+)",
849 process.stderr,
850 ):
851 msg = f"Cannot send {src_dataset}: {match.group(3)}"
852 raise RetryableError(display_msg="zfs send") from RuntimeError(msg)
855def _clear_resumable_recv_state(job: Job, dst_dataset: str) -> bool:
856 p, log = job.params, job.params.log
857 log.warning(p.dry("Aborting an interrupted zfs receive -s, deleting partially received state: %s"), dst_dataset)
858 cmd: list[str] = p.split_args(f"{p.dst.sudo} {p.zfs_program} receive -A", dst_dataset)
859 job.try_ssh_command(p.dst, LOG_TRACE, is_dry=p.dry_run, print_stdout=True, cmd=cmd)
860 log.log(LOG_TRACE, p.dry("Done Aborting an interrupted zfs receive -s: %s"), dst_dataset)
861 return True
864def _clear_resumable_recv_state_if_necessary(job: Job, dst_dataset: str, stderr: str) -> bool:
865 """Deletes leftover ZFS resume token state on the receiving dataset if necessary to continue operations.
867 To make resumable ZFS receive a reliable feature, we cope with the following ZFS facts:
868 - A failed `zfs receive -s` prohibits the following subsequent operations, until the situation is explicitly resolved
869 via a successful subsequent `zfs receive`, or cleared via `zfs receive -A`:
870 - `zfs receive` without the resumable receive token (`zfs send -t <token>` is now required)
871 - `zfs destroy <snapshot>`
872 - `zfs rollback`
873 - `zfs send -t` does not support sending more than a single snapshot; e.g. https://github.com/openzfs/zfs/issues/16764
874 - A stale receive token prohibits subsequent `zfs send -t` if not handled (meanwhile, state changed on src or dst).
875 - `zfs receive -A` fails if the receiving dataset has no ZFS resume token (anymore).
876 """
878 # No i18n needed here. OpenZFS ships no translation catalogs, so gettext falls back to English msgids and locale settings
879 # have no effect. If translations ever appear, revisit this or inject LC_ALL=C.
881 # "cannot resume send: 'wb_src/tmp/src@s1' is no longer the same snapshot used in the initial send"
882 # "cannot resume send: 'wb_src/tmp/src@s1' used in the initial send no longer exists"
883 # "cannot resume send: incremental source 0xa000000000000000 no longer exists"
884 if "cannot resume send" in stderr and (
885 "is no longer the same snapshot used in the initial send" in stderr
886 or "used in the initial send no longer exists" in stderr
887 or re.search(r"incremental source [0-9a-fx]+ no longer exists", stderr)
888 ):
889 return _clear_resumable_recv_state(job, dst_dataset)
891 # "cannot receive resume stream: incompatible embedded data stream feature with encrypted receive."
892 # see https://github.com/openzfs/zfs/issues/12480
893 # 'cannot receive new filesystem stream: destination xx contains partially-complete state from "zfs receive -s"'
894 # this indicates that --no-resume-recv detects that dst contains a previously interrupted recv -s
895 elif "cannot receive" in stderr and (
896 "cannot receive resume stream: incompatible embedded data stream feature with encrypted receive" in stderr
897 or 'contains partially-complete state from "zfs receive -s"' in stderr
898 ):
899 return _clear_resumable_recv_state(job, dst_dataset)
901 elif ( # this signals normal behavior on interrupt of 'zfs receive -s' if running without --no-resume-recv
902 "cannot receive new filesystem stream: checksum mismatch or incomplete stream" in stderr
903 and "Partially received snapshot is saved" in stderr
904 ):
905 return True
907 # "cannot destroy 'wb_dest/tmp/dst@s1': snapshot has dependent clones ... use '-R' to destroy the following
908 # datasets: wb_dest/tmp/dst/%recv" # see https://github.com/openzfs/zfs/issues/10439#issuecomment-642774560
909 # This msg indicates a failed 'zfs destroy' via --delete-dst-snapshots. This "clone" is caused by a previously
910 # interrupted 'zfs receive -s'. The fix used here is to delete the partially received state of said
911 # 'zfs receive -s' via 'zfs receive -A', followed by an automatic retry, which will now succeed to delete the
912 # snapshot without user intervention.
913 elif (
914 "cannot destroy" in stderr
915 and "snapshot has dependent clone" in stderr
916 and "use '-R' to destroy the following dataset" in stderr
917 and f"\n{dst_dataset}/%recv\n" in stderr
918 ):
919 return _clear_resumable_recv_state(job, dst_dataset)
921 # Same cause as above, except that this error can occur during 'zfs rollback'
922 # Also see https://github.com/openzfs/zfs/blob/master/cmd/zfs/zfs_main.c
923 elif (
924 "cannot rollback to" in stderr
925 and "clones of previous snapshots exist" in stderr
926 and "use '-R' to force deletion of the following clones and dependents" in stderr
927 and f"\n{dst_dataset}/%recv\n" in stderr
928 ):
929 return _clear_resumable_recv_state(job, dst_dataset)
931 return False
934def _recv_resume_token(job: Job, dst_dataset: str) -> tuple[str | None, list[str], list[str]]:
935 """Gets recv_resume_token ZFS property from dst_dataset and returns corresponding opts to use for send+recv."""
936 p, log = job.params, job.params.log
937 if not p.resume_recv:
938 return None, [], []
939 warning: str | None = None
940 if not is_zpool_feature_enabled_or_active(p, p.dst, "feature@extensible_dataset"):
941 warning = "not available on destination dataset"
942 elif not p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_1_0, "dst"):
943 warning = "unreliable as zfs version is too old" # e.g. zfs-0.8.3 "internal error: Unknown error 1040"
944 if warning:
945 log.warning(f"ZFS receive resume feature is {warning}. Falling back to --no-resume-recv: %s", dst_dataset)
946 return None, [], []
947 recv_resume_token: str | None = None
948 send_resume_opts: list[str] = []
949 if job.dst_dataset_exists[dst_dataset]:
950 cmd: list[str] = p.split_args(f"{p.zfs_program} get -Hp -o value -s none receive_resume_token", dst_dataset)
951 job.maybe_inject_delete(p.dst, dataset=dst_dataset, delete_trigger="zfs_get_recv_resume_token")
952 try:
953 recv_resume_token = job.run_ssh_command(p.dst, LOG_TRACE, cmd=cmd).rstrip()
954 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
955 raise RetryableError(display_msg="zfs get receive_resume_token") from e
956 if recv_resume_token == "-" or not recv_resume_token: # noqa: S105
957 recv_resume_token = None
958 else:
959 send_resume_opts += ["-n"] if p.dry_run_no_send else []
960 send_resume_opts += ["-v"] if p.verbose_zfs else []
961 send_resume_opts += ["-t", recv_resume_token]
962 recv_resume_opts = ["-s"]
963 return recv_resume_token, send_resume_opts, recv_resume_opts
966def _decode_resume_token(
967 job: Job, recv_resume_token: str, src_dataset: str, dst_dataset: str, included_src_guids: Mapping[str, str]
968) -> str:
969 """Return the token's source snapshot after validating identity, GUID/name and effective raw mode."""
970 p, log = job.params, job.params.log
971 decode_cmd: list[str] = p.split_args(f"{p.src.sudo} {p.zfs_program} send -n -v -t", recv_resume_token)
972 try:
973 decoded_resume_token: str = job.run_ssh_command(p.src, LOG_DEBUG, cmd=decode_cmd)
974 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
975 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
976 retry_immediately_once: bool = _clear_resumable_recv_state_if_necessary(job, dst_dataset, stderr)
977 # op isn't idempotent so retries regather current state from the start of replicate_dataset()
978 raise RetryableError(display_msg="zfs send -t", retry_immediately_once=retry_immediately_once) from e
980 name_match: re.Match[str] | None = re.search(r"(?m)^\s+toname\s=\s(.+)$", decoded_resume_token)
981 guid_match: re.Match[str] | None = re.search(r"(?m)^\s+toguid\s=\s(.+)$", decoded_resume_token)
982 rawok_match: re.Match[str] | None = re.search(r"(?m)^\s+rawok\s=", decoded_resume_token)
983 assert name_match is not None and guid_match is not None, f"Cannot parse {recv_resume_token}, {decoded_resume_token}"
984 decoded_src_snapshot: str = name_match.group(1)
985 decoded_src_guid: str = str(int(guid_match.group(1).strip(), base=16))
986 log.log(LOG_TRACE, f"decoded recv_resume_token src snapshot: {decoded_src_snapshot}, guid: {decoded_src_guid}")
987 assert "@" in decoded_src_snapshot, decoded_src_snapshot
988 decoded_src_dataset, decoded_src_tag = decoded_src_snapshot.split("@", 1)
989 assert decoded_src_tag
990 if decoded_src_dataset != src_dataset:
991 _clear_resumable_recv_state(job, dst_dataset)
992 msg = "because zfs receive resume token is stale"
993 raise RetryableError(display_msg=msg, retry_immediately_once=True) from RuntimeError(msg)
994 if included_src_guids.get(decoded_src_guid) != decoded_src_snapshot:
995 _clear_resumable_recv_state(job, dst_dataset)
996 msg = "because zfs receive resume token does not match a selected source snapshot GUID and name"
997 raise RetryableError(display_msg=msg, retry_immediately_once=True) from RuntimeError(msg)
999 def _is_raw_zfs_send(send_opts: Iterable[str]) -> bool:
1000 return any(opt == "--raw" or ("w" in opt and opt.startswith("-") and not opt.startswith("--")) for opt in send_opts)
1002 is_raw_token: bool = bool(rawok_match)
1003 is_raw_send: bool = _is_raw_zfs_send(p.curr_zfs_send_program_opts)
1004 is_raw_mode_conflict: bool = is_raw_token != is_raw_send
1005 if is_raw_mode_conflict and not is_raw_token:
1006 # `--raw` does not produce a raw stream for an unencrypted source, so a token without `rawok` remains compatible.
1007 cmd: list[str] = p.split_args(f"{p.zfs_program} get -Hp -o value -s none encryptionroot", src_dataset)
1008 encryptionroot: str = (job.try_ssh_command(p.src, LOG_TRACE, cmd=cmd, exists=False) or "").rstrip("\n")
1009 assert encryptionroot
1010 is_raw_mode_conflict = encryptionroot != "-" # encryptionroot != "-" indicates an encrypted dataset
1011 if is_raw_mode_conflict:
1012 _clear_resumable_recv_state(job, dst_dataset)
1013 if p.dry_run:
1014 die("Cannot clear the ZFS receive resume token because --dryrun never modifies state.")
1015 msg = f"as zfs receive resume token raw mode conflicts with --zfs-send-program-opts: {p.curr_zfs_send_program_opts}"
1016 raise RetryableError(display_msg=msg, retry_immediately_once=True) from RuntimeError(msg)
1017 return decoded_src_snapshot
1020def _mbuffer_cmd(p: Params, loc: str, recordsize: int) -> str:
1021 """Use mbuffer in the network pipe between 'zfs send' and 'zfs receive' to smooth out the rate of data flow and prevent
1022 bottlenecks caused by network latency or speed fluctuation."""
1023 recordsize = max(recordsize, 2 * 1024 * 1024)
1024 mbuffer_program_opts: list[str] = [p.mbuffer_program, "-s", str(recordsize)] + p.mbuffer_program_opts
1025 if p.bwlimit:
1026 mbuffer_program_opts += ["-R" if loc == "src" else "-r", p.bwlimit.upper()]
1027 return shlex.join(mbuffer_program_opts)
1030def _compress_cmd(p: Params) -> str:
1031 """Use compression (e.g. zstd) in the network pipe between 'zfs send' and 'zfs receive' to reduce network bottlenecks."""
1032 return shlex.join([p.compression_program, "-c"] + p.compression_program_opts)
1035def _decompress_cmd(p: Params) -> str:
1036 """Formats the decompression half of an enabled compression pair."""
1037 return shlex.join([p.compression_program, "-dc"])
1040_WORKER_THREAD_NUMBER_REGEX: Final[re.Pattern[str]] = re.compile(r"ThreadPoolExecutor-\d+_(\d+)")
1043def _pv_cmd(job: Job, size_estimate_bytes: int, size_estimate_human: str, disable_progress_bar: bool = False) -> str:
1044 """`pv` monitors the progress of data transfer from 'zfs send' to 'zfs receive'; Progress can be viewed via
1045 `tail -f $pv_log_file` aka `tail -f ~/bzfs-logs/current/current.pv` or similar."""
1046 p = job.params
1047 size: str = f"--size={size_estimate_bytes}"
1048 if disable_progress_bar or p.no_estimate_send_size:
1049 size = ""
1050 pv_log_file: str = p.log_params.pv_log_file
1051 thread_name: str = threading.current_thread().name
1052 if match := _WORKER_THREAD_NUMBER_REGEX.fullmatch(thread_name):
1053 worker = int(match.group(1))
1054 if worker > 0:
1055 pv_log_file += PV_FILE_THREAD_SEPARATOR + f"{worker:04}"
1056 if job.is_first_replication_task.get_and_set(False):
1057 if not p.log_params.quiet:
1058 job.progress_reporter.start()
1059 job.replication_start_time_nanos = time.monotonic_ns()
1060 if not p.log_params.quiet:
1061 with open_nofollow(pv_log_file, mode="a", encoding="utf-8", perm=FILE_PERMISSIONS) as fd:
1062 fd.write("\n") # mark start of new stream so ProgressReporter can reliably reset bytes_in_flight
1063 job.progress_reporter.enqueue_pv_log_file(pv_log_file)
1064 pv_program_opts: list[str] = [p.pv_program] + p.pv_program_opts
1065 if job.progress_update_intervals is not None: # for testing
1066 pv_program_opts += [f"--interval={job.progress_update_intervals[0]}"]
1067 pv_program_opts += ["--force", f"--name={size_estimate_human}"]
1068 pv_program_opts += [size] if size else []
1069 return f"LC_ALL=C {shlex.join(pv_program_opts)} 2>> {shlex.quote(pv_log_file)}"
1072def is_inconsistent(job: Job, remote: Remote, dataset: str) -> bool:
1073 """Returns the `inconsistent` ZFS property of the given dataset."""
1074 p = job.params
1075 cmd: list[str] = p.split_args(f"{p.zfs_program} list -t filesystem,volume -Hp -o inconsistent", dataset)
1076 try:
1077 return job.run_ssh_command(remote, LOG_TRACE, cmd=cmd).rstrip() != "0"
1078 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1079 raise RetryableError(display_msg="zfs list -o inconsistent") from e
1082def delete_snapshots(job: Job, remote: Remote, dataset: str, snapshot_tags: list[str]) -> None:
1083 """Deletes snapshots in manageable batches on the specified remote."""
1084 if len(snapshot_tags) == 0:
1085 return
1086 p, log = job.params, job.params.log
1087 log.info(p.dry(f"Deleting {len(snapshot_tags)} snapshots within %s: %s"), dataset, snapshot_tags)
1088 # delete snapshots in batches without creating a command line that's too big for the OS to handle
1089 run_ssh_cmd_batched(
1090 job,
1091 remote,
1092 _delete_snapshot_cmd(p, remote, dataset + "@"),
1093 snapshot_tags,
1094 lambda batch: _delete_snapshot(job, remote, dataset, dataset + "@" + ",".join(batch)),
1095 max_batch_items=job.params.max_snapshots_per_minibatch_on_delete_snaps,
1096 sep=",",
1097 )
1100def _delete_snapshot(job: Job, r: Remote, dataset: str, snapshots_to_delete: str) -> None:
1101 """Runs zfs destroy for a comma-separated snapshot list."""
1102 p = job.params
1103 cmd: list[str] = _delete_snapshot_cmd(p, r, snapshots_to_delete)
1104 is_dry: bool = False # False is safe because we're using the 'zfs destroy -n' flag
1105 try:
1106 job.maybe_inject_error(cmd=cmd, error_trigger="zfs_delete_snapshot")
1107 job.run_ssh_command(r, LOG_DEBUG, is_dry=is_dry, print_stdout=True, cmd=cmd)
1108 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1109 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
1110 retry_immediately_once: bool = _clear_resumable_recv_state_if_necessary(job, dataset, stderr)
1111 # op isn't idempotent so retries regather current state from the start of delete_destination_snapshots() or similar
1112 raise RetryableError(display_msg="zfs destroy snapshot", retry_immediately_once=retry_immediately_once) from e
1115def _delete_snapshot_cmd(p: Params, r: Remote, snapshots_to_delete: str) -> list[str]:
1116 """Builds zfs destroy command for given snapshots."""
1117 return p.split_args(
1118 f"{r.sudo} {p.zfs_program} destroy", p.force_hard, p.verbose_destroy, p.dry_run_destroy, snapshots_to_delete
1119 )
1122def delete_bookmarks(job: Job, remote: Remote, dataset: str, snapshot_tags: list[str], loglevel: int = logging.INFO) -> None:
1123 """Removes bookmarks individually since zfs lacks batch deletion."""
1124 if len(snapshot_tags) == 0:
1125 return
1126 # Unfortunately ZFS has no syntax yet to delete multiple bookmarks in a single CLI invocation
1127 p, log = job.params, job.params.log
1128 log.log(
1129 loglevel,
1130 p.dry(f"Deleting {len(snapshot_tags)} bookmarks within %s: %s"),
1131 dataset,
1132 dataset + "#" + ",".join(snapshot_tags),
1133 )
1134 cmd: list[str] = p.split_args(f"{remote.sudo} {p.zfs_program} destroy")
1135 run_ssh_cmd_parallel(
1136 job,
1137 remote,
1138 [(cmd, (f"{dataset}#{snapshot_tag}" for snapshot_tag in snapshot_tags))],
1139 lambda _cmd, batch: job.try_ssh_command(remote, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=_cmd + batch),
1140 max_batch_items=1,
1141 )
1144def delete_datasets(job: Job, remote: Remote, datasets: Iterable[str]) -> None:
1145 """Deletes the given datasets via zfs destroy -r on the given remote."""
1146 # Impl is batch optimized to minimize CLI + network roundtrips: only need to run zfs destroy if previously
1147 # destroyed dataset (within sorted datasets) is not a prefix (aka ancestor) of current dataset
1148 p, log = job.params, job.params.log
1149 last_deleted_dataset: str = DONT_SKIP_DATASET
1150 for dataset in sort_datasets(datasets):
1151 if is_descendant(dataset, of_root_dataset=last_deleted_dataset):
1152 continue
1153 log.info(p.dry("Deleting dataset tree: %s"), f"{dataset} ...")
1154 cmd: list[str] = p.split_args(
1155 f"{remote.sudo} {p.zfs_program} destroy -r {p.force_unmount} {p.force_hard} {p.verbose_destroy}",
1156 p.dry_run_destroy,
1157 dataset,
1158 )
1159 is_dry: bool = False # False is safe because we're using the 'zfs destroy -n' flag
1160 job.run_ssh_command(remote, LOG_DEBUG, is_dry=is_dry, print_stdout=True, cmd=cmd)
1161 last_deleted_dataset = dataset
1164def _create_zfs_filesystem(job: Job, filesystem: str) -> None:
1165 """Creates destination filesystem hierarchies without mounting them."""
1166 # zfs create -p -u $filesystem
1167 # To ensure the filesystems that we create do not get mounted, we apply a separate 'zfs create -p -u'
1168 # invocation for each non-existing ancestor. This is because a single 'zfs create -p -u' applies the '-u'
1169 # part only to the immediate filesystem, rather than to the not-yet existing ancestors.
1170 p = job.params
1171 parent: str = ""
1172 no_mount: str = "-u" if p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_1_0, "dst") else ""
1173 for component in filesystem.split("/"):
1174 parent += component
1175 if not job.dst_dataset_exists[parent]:
1176 cmd: list[str] = p.split_args(f"{p.dst.sudo} {p.zfs_program} create -p", no_mount, parent)
1177 try:
1178 job.maybe_inject_error(cmd=cmd, error_trigger="zfs_create_filesystem")
1179 job.run_ssh_command(p.dst, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd)
1180 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1181 # ignore harmless error caused by 'zfs create' without the -u flag, or by dataset already existing
1182 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
1183 is_harmless: bool = isinstance(e, subprocess.CalledProcessError) and (
1184 "filesystem successfully created, but it may only be mounted by root" in stderr
1185 or "filesystem successfully created, but not mounted" in stderr # SolarisZFS
1186 or "dataset already exists" in stderr
1187 or "filesystem already exists" in stderr # SolarisZFS?
1188 )
1189 if not is_harmless:
1190 raise RetryableError(display_msg="zfs create") from e
1191 if not p.dry_run:
1192 job.dst_dataset_exists[parent] = True
1193 parent += "/"
1196def _create_zfs_bookmarks(
1197 job: Job, remote: Remote, snapshots: list[str], bookmarks: list[str], *, expected_guids: list[str]
1198) -> None:
1199 """Creates bookmarks with the given names for the given snapshots (or bookmarks), using the 'zfs bookmark' CLI;
1200 accepts an existing target bookmark only when its GUID matches the expected GUID."""
1201 # Unfortunately ZFS has no syntax yet to create multiple bookmarks in a single CLI invocation
1202 p = job.params
1203 assert len(snapshots) == len(bookmarks)
1204 assert len(snapshots) == len(expected_guids)
1205 bookmark_dict: dict[str, tuple[str, str]] = {}
1206 for i, snapshot in enumerate(snapshots):
1207 bookmark_dict[snapshot] = (bookmarks[i], expected_guids[i])
1208 assert len(snapshots) == len(bookmark_dict)
1210 def create_zfs_bookmark(cmd: list[str]) -> None:
1211 snapshot = cmd[-1]
1212 bookmark, expected_guid = bookmark_dict[snapshot]
1213 try:
1214 job.run_ssh_command(remote, LOG_DEBUG, is_dry=p.dry_run, print_stderr=False, cmd=cmd + [bookmark])
1215 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1216 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
1217 if ": bookmark exists" in stderr:
1218 # verify value of actual bookmark GUID is as expected
1219 query_cmd: list[str] = p.split_args(f"{p.zfs_program} list -t bookmark -Hp -o guid", bookmark)
1220 try:
1221 job.maybe_inject_error(cmd=query_cmd, error_trigger="zfs_list_bookmark_guid")
1222 actual_guid: str = job.run_ssh_command(remote, LOG_DEBUG, cmd=query_cmd).rstrip()
1223 except (subprocess.CalledProcessError, UnicodeDecodeError) as query_error:
1224 raise RetryableError(display_msg="zfs list -t bookmark") from query_error
1225 if actual_guid == expected_guid:
1226 return # harmless
1227 die(
1228 f"Conflict: Cannot create bookmark {bookmark!r} with expected GUID {expected_guid!r} from {snapshot!r} "
1229 f"because the bookmark already exists with a different GUID {actual_guid!r}"
1230 )
1231 xprint(p.log, stderr, file=sys.stderr, end="")
1232 raise RetryableError(display_msg="zfs bookmark") from e
1234 cmd: list[str] = p.split_args(f"{remote.sudo} {p.zfs_program} bookmark")
1235 run_ssh_cmd_parallel(
1236 job, remote, [(cmd, snapshots)], lambda _cmd, batch: create_zfs_bookmark(_cmd + batch), max_batch_items=1
1237 )
1240def _estimate_send_size(job: Job, remote: Remote, dst_dataset: str, recv_resume_token: str | None, *items: str) -> int:
1241 """Estimates num bytes to transfer via 'zfs send -nvP'; Thread-safe."""
1242 p = job.params
1243 if p.no_estimate_send_size:
1244 return 0
1245 zfs_send_program_opts: list[str] = p.curr_zfs_send_resume_opts if recv_resume_token else p.curr_zfs_send_program_opts
1246 zfs_send_program_opts = ["--parsable" if opt == "-P" else opt for opt in zfs_send_program_opts]
1247 zfs_send_program_opts = append_if_absent(zfs_send_program_opts, "-v", "-n", "--parsable")
1248 if recv_resume_token:
1249 zfs_send_program_opts += ["-t", recv_resume_token]
1250 items = ()
1251 cmd: list[str] = p.split_args(f"{remote.sudo} {p.zfs_program} send", zfs_send_program_opts, items)
1252 try:
1253 lines: str | None = job.try_ssh_command(remote, LOG_TRACE, cmd=cmd)
1254 except RetryableError as retryable_error:
1255 assert retryable_error.__cause__ is not None
1256 if recv_resume_token:
1257 e = retryable_error.__cause__
1258 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else ""
1259 retryable_error.retry_immediately_once = _clear_resumable_recv_state_if_necessary(job, dst_dataset, stderr)
1260 # op isn't idempotent so retries regather current state from the start of replicate_dataset()
1261 raise
1262 if lines is None: 1262 ↛ 1263line 1262 didn't jump to line 1263 because the condition on line 1262 was never true
1263 return 0 # src dataset or snapshot has been deleted by third party
1264 size: str = lines.splitlines()[-1]
1265 assert size.startswith("size")
1266 return int(size[size.index("\t") + 1 :])
1269def _estimate_send_sizes_in_parallel(
1270 job: Job,
1271 r: Remote,
1272 dst_dataset: str,
1273 recv_resume_token: str | None,
1274 steps_todo: list[tuple[str, str, str, list[str]]],
1275) -> list[int]:
1276 """Estimates num bytes to transfer for multiple send steps; in parallel to reduce latency."""
1277 p = job.params
1278 if p.no_estimate_send_size:
1279 return [0 for _ in steps_todo]
1281 def iterator_builder(executor: Executor) -> Iterable[Iterable[Future[int]]]:
1282 resume_token: str | None = recv_resume_token
1283 return [
1284 (
1285 executor.submit(
1286 _estimate_send_size, job, r, dst_dataset, resume_token if i == 0 else None, incr_flag, from_snap, to_snap
1287 )
1288 for i, (incr_flag, from_snap, to_snap, _to_snapshots) in enumerate(steps_todo)
1289 ) # advancing the lazy on-demand Python Generator submits the next task and yields the corresponding Future
1290 ]
1292 max_workers: int = min(len(steps_todo), job.max_workers[r.location])
1293 return list(
1294 parallel_iterator(
1295 iterator_builder, max_workers=max_workers, ordered=True, is_terminated=job.termination_event.is_set
1296 )
1297 )
1300def _zfs_set(job: Job, properties: list[str], remote: Remote, dataset: str) -> None:
1301 """Applies the given property key=value pairs via 'zfs set' CLI to the given dataset on the given remote."""
1302 p = job.params
1303 if len(properties) == 0:
1304 return
1305 # set properties in batches without creating a command line that's too big for the OS to handle
1306 cmd: list[str] = p.split_args(f"{remote.sudo} {p.zfs_program} set")
1307 run_ssh_cmd_batched(
1308 job,
1309 remote,
1310 cmd + [dataset],
1311 properties,
1312 lambda batch: job.run_ssh_command(
1313 remote, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd + batch + [dataset]
1314 ),
1315 max_batch_items=2**29,
1316 )
1319def _zfs_get(
1320 job: Job,
1321 remote: Remote,
1322 dataset: str,
1323 sources: str,
1324 output_columns: str,
1325 propnames: str,
1326 splitlines: bool,
1327 props_cache: dict[tuple[str, ...], dict[str, str | None]],
1328 refresh: bool = False,
1329) -> dict[str, str | None]:
1330 """Returns the results of 'zfs get' CLI on the given dataset on the given remote."""
1331 assert dataset
1332 assert sources
1333 assert output_columns
1334 if not propnames:
1335 return {}
1336 p = job.params
1337 cache_key: tuple[str, ...] = (remote.location, dataset, sources, output_columns, propnames)
1338 if refresh:
1339 props_cache.pop(cache_key, None)
1340 props: dict[str, str | None] | None = props_cache.get(cache_key)
1341 if props is None:
1342 cmd: list[str] = p.split_args(f"{p.zfs_program} get -Hp -o {output_columns} -s {sources} {propnames}", dataset)
1343 lines: str = job.run_ssh_command(remote, LOG_TRACE, cmd=cmd)
1344 is_name_value_pair: bool = "," in output_columns
1345 props = {}
1346 # if not splitlines: omit single trailing newline that was appended by 'zfs get' CLI
1347 assert splitlines or len(lines) == 0 or lines[-1] == "\n"
1348 for line in lines.splitlines() if splitlines else [lines[0:-1]]:
1349 if is_name_value_pair:
1350 propname, propvalue = line.split("\t", 1)
1351 props[propname] = propvalue
1352 else:
1353 props[line] = None
1354 props_cache[cache_key] = props
1355 return props
1358def _incremental_send_steps_wrapper(
1359 p: Params, src_snapshots: list[str], src_guids: list[str], included_guids: set[str], is_resume: bool
1360) -> list[tuple[str, str, str, list[str]]]:
1361 """Returns incremental send steps, optionally converting -I to -i."""
1362 force_convert_I_to_i: bool = not getenv_bool("no_force_convert_I_to_i", True) # noqa: N806
1363 # force_convert_I_to_i == True implies that:
1364 # Force convert 'zfs send -I' to a series of 'zfs send -i', for example as a workaround for
1365 # zfs issue https://github.com/openzfs/zfs/issues/16394
1366 return incremental_send_steps(src_snapshots, src_guids, included_guids, is_resume, force_convert_I_to_i)
1369def _add_recv_property_options(
1370 job: Job, full_send: bool, recv_opts: list[str], dataset: str, cache: dict[tuple[str, ...], dict[str, str | None]]
1371) -> tuple[list[str], list[str]]:
1372 """Reads the ZFS properties of the given src dataset; Appends zfs recv -o and -x values to a recv_opts copy according to
1373 CLI params, and returns properties to explicitly set on the dst dataset after 'zfs receive' completes successfully."""
1374 is_volume: bool = job.src_properties[dataset].recordsize < 0
1375 p = job.params
1376 recv_opts = recv_opts.copy()
1377 set_opts: list[str] = []
1378 x_names: list[str] = p.zfs_recv_x_names
1379 x_names_set: set[str] = set(x_names)
1380 ox_names: set[str] = p.zfs_recv_ox_names.copy()
1381 if p.is_program_available(ZFS_VERSION_IS_AT_LEAST_2_2_0, p.dst.location):
1382 # workaround for https://github.com/openzfs/zfs/commit/b0269cd8ced242e66afc4fa856d62be29bb5a4ff
1383 # 'zfs recv -x foo' on zfs < 2.2 errors out if the 'foo' property isn't contained in the send stream
1384 for propname in x_names:
1385 recv_opts.append("-x")
1386 recv_opts.append(propname)
1387 ox_names.update(x_names) # union
1388 for config in [p.zfs_recv_o_config, p.zfs_recv_x_config, p.zfs_set_config]:
1389 if len(config.include_regexes) == 0:
1390 continue # this is the default - it's an instant noop
1391 if (full_send and "full" in config.targets) or (not full_send and "incremental" in config.targets):
1392 # 'zfs get' uses newline as record separator and tab as separator between output columns. A ZFS user property
1393 # may contain newline and tab characters (indeed anything). Together, this means that there is no reliable
1394 # way to determine where a record ends and the next record starts when listing multiple arbitrary records in
1395 # a single 'zfs get' call. Therefore, here we use a separate 'zfs get' call for each ZFS user property.
1396 # TODO: perf: on zfs >= 2.3 use json via zfs get -j to safely merge all zfs gets into one 'zfs get' call
1397 try:
1398 props_any: dict = _zfs_get(job, p.src, dataset, config.sources, "property", "all", True, cache)
1399 props_filtered: dict = filter_properties(p, props_any, config.include_regexes, config.exclude_regexes)
1400 user_propnames: list[str] = [name for name in props_filtered if ":" in name]
1401 sys_propnames: str = ",".join(name for name in props_filtered if ":" not in name)
1402 props: dict = _zfs_get(job, p.src, dataset, config.sources, "property,value", sys_propnames, True, cache)
1403 for propnames in user_propnames:
1404 props.update(_zfs_get(job, p.src, dataset, config.sources, "property,value", propnames, False, cache))
1405 except (subprocess.CalledProcessError, UnicodeDecodeError) as e:
1406 raise RetryableError(display_msg="zfs get") from e
1407 for propname in sorted(props.keys()):
1408 if config is p.zfs_recv_o_config:
1409 if not (propname in ox_names or propname in x_names_set):
1410 recv_opts.append("-o")
1411 recv_opts.append(f"{propname}={props[propname]}")
1412 ox_names.add(propname)
1413 elif config is p.zfs_recv_x_config:
1414 if propname not in ox_names:
1415 recv_opts.append("-x")
1416 recv_opts.append(propname)
1417 ox_names.add(propname)
1418 else:
1419 assert config is p.zfs_set_config
1420 set_opts.append(f"{propname}={props[propname]}")
1421 recv_opts = _sanitize_recv_opts_for_dataset_type(recv_opts, is_volume=is_volume)
1422 return recv_opts, set_opts
1425_ZFS_RECV_PROPS_REJECTED_ON_ANY_DATASET: Final[frozenset[str]] = frozenset(
1426 ["casesensitivity", "normalization", "utf8only", "volblocksize", "volsize"]
1427)
1428_ZFS_RECV_PROPS_REJECTED_ON_FILESYSTEM: Final[Mapping[str, frozenset[str]]] = {
1429 "-x": _ZFS_RECV_PROPS_REJECTED_ON_ANY_DATASET,
1430 "-o": _ZFS_RECV_PROPS_REJECTED_ON_ANY_DATASET,
1431}
1432_ZFS_RECV_PROPS_REJECTED_ON_ZVOL: Final[Mapping[str, frozenset[str]]] = {
1433 "-x": _ZFS_RECV_PROPS_REJECTED_ON_ANY_DATASET,
1434 "-o": frozenset(
1435 {
1436 # see https://github.com/openzfs/zfs/blob/master/module/zcommon/zfs_prop.c
1437 "aclinherit",
1438 "aclmode",
1439 "acltype",
1440 "atime",
1441 "canmount",
1442 "devices",
1443 "mountpoint",
1444 "overlay",
1445 "recordsize",
1446 "relatime",
1447 "sharenfs",
1448 "sharesmb",
1449 "snapdir",
1450 "xattr",
1451 }.union(_ZFS_RECV_PROPS_REJECTED_ON_ANY_DATASET)
1452 ),
1453}
1456def _sanitize_recv_opts_for_dataset_type(recv_opts: list[str], *, is_volume: bool) -> list[str]:
1457 """Drops `zfs receive` `-o/-x` properties that `zfs receive` rejects as an error for this dataset type. Keeps properties
1458 for which `zfs receive` emits only (harmless) warnings.
1460 For example:
1461 drop -o canmount=<value> on zvols
1462 drop -o recordsize=<value> on zvols
1463 drop -o mountpoint=<value> on zvols
1464 drop -o/-x casesensitivity on zvols and filesystems
1465 drop -o/-x volsize on zvols and filesystems
1466 keep -o/-x volmode on zvols and filesystems
1467 """
1468 if is_volume:
1469 inapplicable_props_dict = _ZFS_RECV_PROPS_REJECTED_ON_ZVOL
1470 else:
1471 inapplicable_props_dict = _ZFS_RECV_PROPS_REJECTED_ON_FILESYSTEM
1473 results: list[str] = []
1474 i = 0
1475 while i < len(recv_opts):
1476 opt: str = recv_opts[i]
1477 i += 1
1478 inapplicable_props: frozenset[str] | None = inapplicable_props_dict.get(opt)
1479 if inapplicable_props is None or i >= len(recv_opts):
1480 results.append(opt)
1481 continue
1482 arg: str = recv_opts[i]
1483 i += 1
1484 propname: str = arg.split("=", 1)[0] if opt == "-o" else arg
1485 if propname not in inapplicable_props: # retain this property on this dataset type
1486 results.append(opt)
1487 results.append(arg)
1488 return results
1491def _check_zfs_dataset_busy(job: Job, remote: Remote, dataset: str, busy_if_send: bool = True) -> bool:
1492 """Decline to start a state changing ZFS operation that is, although harmless, likely to collide with another currently
1493 running process. Instead, retry the operation later, after some delay. For example, decline to start a 'zfs receive' into
1494 a destination dataset if another process is already running another 'zfs receive' into the same destination dataset, as
1495 ZFS would reject any such attempt. However, it's actually fine to run an incremental 'zfs receive' into a dataset in
1496 parallel with a 'zfs send' out of the very same dataset. This also helps daisy chain use cases where A replicates to B,
1497 and B replicates to C.
1499 _check_zfs_dataset_busy() offers no guarantees, it merely proactively avoids likely collisions. In other words, even if
1500 the process check below passes there is no guarantee that the destination dataset won't be busy by the time we actually
1501 execute the 'zfs send' operation. In such an event ZFS will reject the operation, we'll detect that, and we'll simply
1502 auto-retry, after some delay. _check_zfs_dataset_busy() can be disabled via --ps-program=-.
1504 TLDR: As is common for long-running operations in distributed systems, we use coordination-free optimistic concurrency
1505 control where the parties simply retry on collision detection (rather than coordinate concurrency via a remote lock
1506 server).
1507 """
1508 p, log = job.params, job.params.log
1509 if not p.is_program_available("ps", remote.location):
1510 return True
1511 cmd: list[str] = p.split_args(f"{p.ps_program} -Ao args")
1512 procs: list[str] = (job.try_ssh_command(remote, LOG_TRACE, cmd=cmd) or "").splitlines()
1513 if job.inject_params.get("is_zfs_dataset_busy", False):
1514 procs += ["sudo -n zfs receive -u -o foo:bar=/baz " + dataset] # for unit testing only
1515 if not _is_zfs_dataset_busy(procs, dataset, busy_if_send=busy_if_send):
1516 return True
1517 op: str = "zfs {receive" + ("|send" if busy_if_send else "") + "} operation"
1518 try:
1519 die(f"Cannot continue now: Destination is already busy with {op} from another process: {dataset}")
1520 except SystemExit as e:
1521 log.warning("%s", e)
1522 raise RetryableError("dst currently busy with zfs mutation op", display_msg="replication") from e
1525_ZFS_DATASET_BUSY_PREFIX: Final[str] = r"(([^ ]*?/)?(sudo|doas)( +-n)? +)?([^ ]*?/)?zfs (receive|recv"
1526_ZFS_DATASET_BUSY_IF_MODS: Final[re.Pattern[str]] = re.compile((_ZFS_DATASET_BUSY_PREFIX + ") .*").replace("(", "(?:"))
1527_ZFS_DATASET_BUSY_IF_SEND: Final[re.Pattern[str]] = re.compile((_ZFS_DATASET_BUSY_PREFIX + "|send) .*").replace("(", "(?:"))
1530def _is_zfs_dataset_busy(procs: list[str], dataset: str, busy_if_send: bool) -> bool:
1531 """Checks if any process list entry indicates ZFS activity on dataset."""
1532 regex: re.Pattern[str] = _ZFS_DATASET_BUSY_IF_SEND if busy_if_send else _ZFS_DATASET_BUSY_IF_MODS
1533 suffix: str = " " + dataset
1534 infix: str = " " + dataset + "@"
1535 return any((proc.endswith(suffix) or infix in proc) and regex.fullmatch(proc) for proc in procs)
1538#############################################################################
1539_TMP_BOOKMARK_PREFIX: Final[str] = ".TMPBZFS."
1540_TMP_BOOKMARK_HASH_PREFIX: Final[str] = "#" + _TMP_BOOKMARK_PREFIX
1543def is_tmp_bookmark(bookmark: str) -> bool:
1544 """Returns whether the given name is a temporary bookmark (from the reserved namespace)."""
1545 return _TMP_BOOKMARK_HASH_PREFIX in bookmark
1548@final
1549class _Continuity:
1550 """Guarantees that a ZFS snapshot can be safely deleted after it has been successfully replicated because we can use a ZFS
1551 bookmark as a common base to continue incremental ZFS replication. This continuity guarantee exists even if the bzfs
1552 process is killed after `zfs send/receive` succeeds but before the post-replication bookmark creation step runs.
1554 The mechanism works as follows: Before replication starts it creates a temporary bookmark for each snapshot that is about
1555 to be replicated. After replication succeeds it promotes (renames) the tmp bookmark to a finalized bookmark. Both the
1556 temporary bookmark as well as the finalized bookmark will be used as a source for incremental ZFS replication. The
1557 mechanism also garbage collects obsolete temporary bookmarks as necessary.
1559 There is no need to eagerly promote/gc on the "already up-to-date" branch (e.g. on retry) as promote/gc is merely
1560 slightly delayed until the next successful replication run.
1562 Name of tmp bookmark = .TMPBZFS.<snapshot_name>.<compact_snapshot_guid>.<dst_dataset_id>
1563 where <snapshot_name> is the part after the '@', and
1564 where <dst_dataset_id> is the compact ZFS GUID of the destination pool followed by a dot followed by a 11-character base64
1565 SHA-256 prefix of the full destination dataset name.
1566 The compact form of a ZFS GUID uses a base64 representation instead of a decimal representation.
1567 Example tmp bookmark for snapshot "bzfs_us-west_2024-11-06_08:30:05_hourly":
1568 ".TMPBZFS.bzfs_us-west_2024-11-06_08:30:05_hourly.ADgurPHaj8k.AAAfaLnRWNQ.mjKQwJEMnjk"
1570 Name of finalized bookmark = same name as its corresponding snapshot except also contains <snapshot_guid> for uniqueness;
1571 it appends the <snapshot_guid> to a suffixless <snapshot_name> or inserts <snapshot_guid> before the last suffix.
1572 Example finalized bookmark for snapshot "bzfs_us-west_2024-11-06_08:30:05_hourly":
1573 "bzfs_us-west_2024-11-06_08:30:05_15813919022682057_hourly"
1575 Note that ZFS guarantees that a ZFS snapshot and its ZFS bookmarks always have the same GUID.
1577 The bookmark namespace starting with ".TMPBZFS." is reserved for internal use. Tmp bookmarks are auto-hidden from other
1578 bzfs operations, e.g. hidden from --delete-dst-snapshots=bookmarks and --compare-snapshot-lists.
1579 """
1581 def __init__(
1582 self,
1583 p: Params,
1584 src_dataset: str,
1585 dst_dataset: str,
1586 raw_src_snapshots_with_guids: list[str],
1587 dst_snapshots_with_guids: list[str],
1588 ) -> None:
1589 self._src_dataset: Final[str] = src_dataset
1590 dst_pool_guid: str = p.zpool_features[p.dst.location][p.dst.pool].get(POOL_GUID, "")
1591 if not dst_pool_guid:
1592 die(
1593 "Cannot create bookmarks to guarantee continuity because the destination zpool GUID could not be detected "
1594 f"for {p.dst.pool!r}. Ensure --zpool-program is enabled and 'zpool' CLI is available on the destination "
1595 "host. Alternatively, consider using --create-bookmarks=none."
1596 )
1597 self._dst_dataset_id: Final[str] = (
1598 self._compact_guid(dst_pool_guid) + "." + self._b64escape(sha256_urlsafe_base64(dst_dataset)[:11])
1599 )
1600 tmp_suffix: str = f".{self._dst_dataset_id}"
1601 self._dst_snapshot_guids: Final[set[str]] = {
1602 guid for guid, name in (line.split("\t", 1) for line in dst_snapshots_with_guids) if "@" in name
1603 }
1604 raw_src_snapshots_components: list[tuple[str, str, bool, bool, bool]] = [
1605 (name, guid, "@" in name, "#" in name, is_tmp_bookmark(name))
1606 for guid, name in (line.split("\t", 1) for line in raw_src_snapshots_with_guids)
1607 ]
1608 self._src_snapshots: Final[Mapping[str, str]] = {
1609 name: guid
1610 for name, guid, is_snapshot, _is_bookmark, _is_tmp_bookmark in raw_src_snapshots_components
1611 if is_snapshot
1612 }
1613 self._tmp_src_bookmarks: Final[dict[str, str]] = {
1614 name: guid
1615 for name, guid, _is_snapshot, is_bookmark, is_tmp_bookmark in raw_src_snapshots_components
1616 if is_bookmark and is_tmp_bookmark and name.endswith(tmp_suffix)
1617 }
1618 self._finalized_src_bookmarks: Final[dict[str, str]] = {
1619 name: guid
1620 for name, guid, _is_snapshot, is_bookmark, is_tmp_bookmark in raw_src_snapshots_components
1621 if is_bookmark and not is_tmp_bookmark
1622 }
1624 def _tmp_bookmark_name_and_guid(self, snapshot: str) -> tuple[str, str]:
1625 """Returns the name and GUID of the tmp bookmark for the given src snapshot."""
1626 assert "@" in snapshot
1627 dataset, snapshot_name = snapshot.split("@", 1)
1628 assert dataset == self._src_dataset
1629 snapshot_guid: str = self._src_snapshots[snapshot]
1630 tbm = f"{dataset}#{_TMP_BOOKMARK_PREFIX}{snapshot_name}.{self._compact_guid(snapshot_guid)}.{self._dst_dataset_id}"
1631 return tbm, snapshot_guid
1633 def _finalized_bookmark_name(self, tmp_bookmark: str) -> str:
1634 """Returns the name of the final bookmark corresponding to the given tmp bookmark."""
1635 assert is_tmp_bookmark(tmp_bookmark), tmp_bookmark
1636 dataset, tag = tmp_bookmark.split("#", 1)
1637 assert dataset == self._src_dataset
1638 prefix_and_snapshot_name, compact_snapshot_guid, _, _ = tag.rsplit(".", 3)
1639 assert prefix_and_snapshot_name.startswith(_TMP_BOOKMARK_PREFIX), (prefix_and_snapshot_name, tmp_bookmark)
1640 snapshot_name = prefix_and_snapshot_name.removeprefix(_TMP_BOOKMARK_PREFIX) # bzfs_us-west_2024-11-06_08:30:05_daily
1641 snapshot_guid: str = self._decimal_guid(compact_snapshot_guid)
1642 if "_" in snapshot_name: # standard name
1643 prefix, suffix = snapshot_name.rsplit("_", 1)
1644 fbm = f"{dataset}#{prefix}_{snapshot_guid}_{suffix}" # bzfs_us-west_2024-11-06_08:30:05_15832610720350849_hourly
1645 else: # non-standard name; e.g. "foo" -> "foo_15832610720350849"
1646 fbm = f"{dataset}#{snapshot_name}_{snapshot_guid}"
1647 return fbm
1649 def create_tmp_bookmarks(self, job: Job, src: Remote, snapshots: list[str]) -> None:
1650 """Adds tmp bookmarks corresponding to the given src snapshots."""
1651 bookmarks: dict[str, str] = dict(self._tmp_bookmark_name_and_guid(snapshot) for snapshot in snapshots)
1652 _create_zfs_bookmarks(job, src, snapshots, list(bookmarks.keys()), expected_guids=list(bookmarks.values()))
1653 for name, guid in bookmarks.items():
1654 self._tmp_src_bookmarks[name] = guid
1656 def mark_snapshots_as_replicated(self, snapshots: list[str]) -> None:
1657 """Records that the given src snapshots have been successfully replicated to the destination dataset."""
1658 for snapshot in snapshots:
1659 snapshot_guid: str = self._src_snapshots[snapshot]
1660 self._dst_snapshot_guids.add(snapshot_guid)
1662 def promote_and_gc_bookmarks(self, job: Job, src: Remote) -> None:
1663 """
1664 For each tmp bookmark that targets the given destination dataset:
1665 Promote aka copy it into a finalized bookmark if a snapshot with the tmp bookmark GUID exists in the dst dataset,
1666 and a corresponding finalized bookmark with the same GUID does not yet exist.
1667 For each tmp bookmark that targets the given destination dataset:
1668 Delete it if a snapshot with the tmp bookmark GUID does not exist in the destination dataset,
1669 or if a corresponding finalized bookmark with the same GUID already exists.
1670 """
1671 tmp_bookmarks_to_delete: list[str] = []
1672 promote_bookmarks: list[tuple[str, str, str]] = []
1673 for tmp_bookmark_name, tmp_bookmark_guid in self._tmp_src_bookmarks.items():
1674 assert tmp_bookmark_name
1675 assert tmp_bookmark_guid
1676 if tmp_bookmark_guid in self._dst_snapshot_guids:
1677 finalized_bookmark_name: str = self._finalized_bookmark_name(tmp_bookmark_name)
1678 finalized_guid: str | None = self._finalized_src_bookmarks.get(finalized_bookmark_name, None)
1679 if finalized_guid is None or tmp_bookmark_guid != finalized_guid:
1680 promote_bookmarks.append((tmp_bookmark_name, tmp_bookmark_guid, finalized_bookmark_name))
1681 tmp_bookmarks_to_delete.append(tmp_bookmark_name)
1682 else:
1683 tmp_bookmarks_to_delete.append(tmp_bookmark_name)
1685 # promote in parallel
1686 _create_zfs_bookmarks(
1687 job,
1688 src,
1689 [tmp_bookmark_name for tmp_bookmark_name, _, _ in promote_bookmarks],
1690 [finalized_bookmark_name for _, _, finalized_bookmark_name in promote_bookmarks],
1691 expected_guids=[tmp_bookmark_guid for _, tmp_bookmark_guid, _ in promote_bookmarks],
1692 )
1693 for _tmp_bookmark_name, tmp_bookmark_guid, finalized_bookmark_name in promote_bookmarks:
1694 self._finalized_src_bookmarks[finalized_bookmark_name] = tmp_bookmark_guid
1696 # gc in parallel
1697 bookmark_tags_to_delete: list[str] = [bookmark.split("#", 1)[1] for bookmark in tmp_bookmarks_to_delete]
1698 delete_bookmarks(job, src, self._src_dataset, bookmark_tags_to_delete, loglevel=LOG_DEBUG)
1699 for bookmark in tmp_bookmarks_to_delete:
1700 self._tmp_src_bookmarks.pop(bookmark)
1702 @staticmethod
1703 def _compact_guid(decimal_guid: str) -> str:
1704 """Converts a 64-bit ZFS GUID from canonical decimal format (20 bytes) to base64 (11 bytes); to shorten it."""
1705 guid_bytes: bytes = base64.urlsafe_b64encode(int(decimal_guid).to_bytes(8, "big"))
1706 return _Continuity._b64escape(guid_bytes.decode().rstrip("="))
1708 @staticmethod
1709 def _decimal_guid(compact_guid: str) -> str:
1710 """Converts an 11-character base64 ZFS GUID back to its canonical decimal representation such that
1711 _decimal_guid(_compact_guid(x)) == x"""
1712 guid_bytes: bytes = base64.urlsafe_b64decode(_Continuity._b64unescape(compact_guid) + "=")
1713 return str(int.from_bytes(guid_bytes, "big"))
1715 @staticmethod
1716 def _b64escape(name: str) -> str:
1717 return name.replace("_", ":")
1719 @staticmethod
1720 def _b64unescape(name: str) -> str:
1721 return name.replace(":", "_")