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