Coverage for bzfs_main / replication.py: 99%

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. 

17 

18For replication of multiple datasets, including recursive replication, see bzfs.py:replicate_datasets(). 

19""" 

20 

21from __future__ import ( 

22 annotations, 

23) 

24import os 

25import re 

26import shlex 

27import subprocess 

28import sys 

29import threading 

30import time 

31from collections.abc import ( 

32 Iterable, 

33 Iterator, 

34) 

35from concurrent.futures import ( 

36 Executor, 

37 Future, 

38) 

39from subprocess import ( 

40 DEVNULL, 

41 PIPE, 

42) 

43from typing import ( 

44 TYPE_CHECKING, 

45 Final, 

46) 

47 

48from bzfs_main.argparse_actions import ( 

49 has_timerange_filter, 

50) 

51from bzfs_main.configuration import ( 

52 is_same_remote, 

53) 

54from bzfs_main.detect import ( 

55 ZFS_VERSION_IS_AT_LEAST_2_1_0, 

56 ZFS_VERSION_IS_AT_LEAST_2_2_0, 

57 are_bookmarks_enabled, 

58 is_zpool_feature_enabled_or_active, 

59) 

60from bzfs_main.filter import ( 

61 filter_properties, 

62 filter_snapshots, 

63) 

64from bzfs_main.incremental_send_steps import ( 

65 incremental_send_steps, 

66) 

67from bzfs_main.parallel_batch_cmd import ( 

68 run_ssh_cmd_batched, 

69 run_ssh_cmd_parallel, 

70) 

71from bzfs_main.progress_reporter import ( 

72 PV_FILE_THREAD_SEPARATOR, 

73) 

74from bzfs_main.util.connection import ( 

75 DEDICATED, 

76 SHARED, 

77 ConnectionPool, 

78 dquote, 

79 squote, 

80 timeout, 

81) 

82from bzfs_main.util.parallel_iterator import ( 

83 parallel_iterator, 

84 run_in_parallel, 

85) 

86from bzfs_main.util.retry import ( 

87 Retry, 

88 RetryableError, 

89) 

90from bzfs_main.util.utils import ( 

91 DONT_SKIP_DATASET, 

92 FILE_PERMISSIONS, 

93 LOG_DEBUG, 

94 LOG_TRACE, 

95 Subprocesses, 

96 append_if_absent, 

97 cut, 

98 die, 

99 getenv_bool, 

100 human_readable_bytes, 

101 is_descendant, 

102 list_formatter, 

103 open_nofollow, 

104 replace_prefix, 

105 stderr_to_str, 

106 xprint, 

107) 

108 

109if TYPE_CHECKING: # pragma: no cover - for type hints only 

110 from bzfs_main.bzfs import ( 

111 Job, 

112 ) 

113 from bzfs_main.configuration import ( 

114 Params, 

115 Remote, 

116 ) 

117 

118 

119# constants: 

120INJECT_DST_PIPE_FAIL_KBYTES: Final[int] = 400 # for testing only 

121_RIGHT_JUST: Final[int] = 7 

122 

123 

124def replicate_dataset(job: Job, src_dataset: str, tid: str, retry: Retry) -> bool: 

125 """Replicates src_dataset to dst_dataset (thread-safe); For replication of multiple datasets, including recursive 

126 replication, see bzfs.py:replicate_datasets().""" 

127 p, log = job.params, job.params.log 

128 src, dst = p.src, p.dst 

129 retry_count: int = retry.count 

130 dst_dataset: str = replace_prefix(src_dataset, old_prefix=src.root_dataset, new_prefix=dst.root_dataset) 

131 log.debug(p.dry(f"{tid} Replicating: %s"), f"{src_dataset} --> {dst_dataset} ...") 

132 

133 list_result: bool | tuple[list[str], list[str], list[str], set[str], str, str] = _list_and_filter_src_and_dst_snapshots( 

134 job, src_dataset, dst_dataset 

135 ) 

136 if isinstance(list_result, bool): 

137 return list_result 

138 ( 

139 basis_src_snapshots_with_guids, 

140 _src_snapshots_with_guids, 

141 dst_snapshots_with_guids, 

142 included_src_guids, 

143 latest_src_snapshot, 

144 oldest_src_snapshot, 

145 ) = list_result 

146 log.debug("latest_src_snapshot: %s", latest_src_snapshot) 

147 latest_dst_snapshot: str = "" 

148 latest_common_src_snapshot: str = "" 

149 done_checking: bool = False 

150 

151 if job.dst_dataset_exists[dst_dataset]: 

152 rollback_result: bool | tuple[str, str, bool] = _rollback_dst_dataset_if_necessary( 

153 job, 

154 dst_dataset, 

155 latest_src_snapshot, 

156 basis_src_snapshots_with_guids, 

157 dst_snapshots_with_guids, 

158 done_checking, 

159 tid, 

160 ) 

161 if isinstance(rollback_result, bool): 

162 return rollback_result 

163 latest_dst_snapshot, latest_common_src_snapshot, done_checking = rollback_result 

164 

165 log.debug("latest_common_src_snapshot: %s", latest_common_src_snapshot) # is a snapshot or bookmark 

166 log.log(LOG_TRACE, "latest_dst_snapshot: %s", latest_dst_snapshot) 

167 props_cache: dict[tuple[str, str, str], dict[str, str | None]] = {} # fresh empty ZFS props cache for each dataset 

168 dry_run_no_send: bool = False 

169 if not latest_common_src_snapshot: 

170 # no common snapshot exists; delete all dst snapshots and perform a full send of the oldest selected src snapshot 

171 full_result: tuple[str, bool, bool, int] = _replicate_dataset_fully( 

172 job, 

173 src_dataset, 

174 dst_dataset, 

175 oldest_src_snapshot, 

176 latest_src_snapshot, 

177 latest_dst_snapshot, 

178 dst_snapshots_with_guids, 

179 props_cache, 

180 dry_run_no_send, 

181 done_checking, 

182 retry_count, 

183 tid, 

184 ) 

185 # we have now created a common snapshot 

186 latest_common_src_snapshot, dry_run_no_send, done_checking, retry_count = full_result 

187 if latest_common_src_snapshot: 

188 # finally, incrementally replicate all selected snapshots from latest common snapshot until latest src snapshot 

189 _replicate_dataset_incrementally( 

190 job, 

191 src_dataset, 

192 dst_dataset, 

193 latest_common_src_snapshot, 

194 latest_src_snapshot, 

195 basis_src_snapshots_with_guids, 

196 included_src_guids, 

197 props_cache, 

198 dry_run_no_send, 

199 done_checking, 

200 retry_count, 

201 tid, 

202 ) 

203 return True 

204 

205 

206def _list_and_filter_src_and_dst_snapshots( 

207 job: Job, src_dataset: str, dst_dataset: str 

208) -> bool | tuple[list[str], list[str], list[str], set[str], str, str]: 

209 """On replication, list and filter src and dst snapshots.""" 

210 p, log = job.params, job.params.log 

211 src, dst = p.src, p.dst 

212 

213 # list GUID and name for dst snapshots, sorted ascending by createtxg (more precise than creation time) 

214 dst_cmd: list[str] = p.split_args(f"{p.zfs_program} list -t snapshot -d 1 -s createtxg -Hp -o guid,name", dst_dataset) 

215 

216 # list GUID and name for src snapshots + bookmarks, primarily sort ascending by transaction group (which is more 

217 # precise than creation time), secondarily sort such that snapshots appear after bookmarks for the same GUID. 

218 # Note: A snapshot and its ZFS bookmarks always have the same GUID, creation time and transaction group. A snapshot 

219 # changes its transaction group but retains its creation time and GUID on 'zfs receive' on another pool, i.e. 

220 # comparing createtxg is only meaningful within a single pool, not across pools from src to dst. Comparing creation 

221 # time remains meaningful across pools from src to dst. Creation time is a UTC Unix time in integer seconds. 

222 # Note that 'zfs create', 'zfs snapshot' and 'zfs bookmark' CLIs enforce that snapshot names must not contain a '#' 

223 # char, bookmark names must not contain a '@' char, and dataset names must not contain a '#' or '@' char. 

224 # GUID and creation time also do not contain a '#' or '@' char. 

225 filter_needs_creation_time: bool = has_timerange_filter(p.snapshot_filters) 

226 types: str = "snapshot,bookmark" if p.use_bookmark and are_bookmarks_enabled(p, src) else "snapshot" 

227 props: str = job.creation_prefix + "creation,guid,name" if filter_needs_creation_time else "guid,name" 

228 src_cmd = p.split_args(f"{p.zfs_program} list -t {types} -s createtxg -s type -d 1 -Hp -o {props}", src_dataset) 

229 job.maybe_inject_delete(src, dataset=src_dataset, delete_trigger="zfs_list_snapshot_src") 

230 src_snapshots_and_bookmarks, dst_snapshots_with_guids_str = run_in_parallel( # list src+dst snapshots in parallel 

231 lambda: job.try_ssh_command(src, LOG_TRACE, cmd=src_cmd), 

232 lambda: job.try_ssh_command(dst, LOG_TRACE, cmd=dst_cmd, error_trigger="zfs_list_snapshot_dst"), 

233 ) 

234 job.dst_dataset_exists[dst_dataset] = dst_snapshots_with_guids_str is not None 

235 dst_snapshots_with_guids: list[str] = (dst_snapshots_with_guids_str or "").splitlines() 

236 if src_snapshots_and_bookmarks is None: 

237 log.warning("Third party deleted source: %s", src_dataset) 

238 return False # src dataset has been deleted by some third party while we're running - nothing to do anymore 

239 src_snapshots_with_guids: list[str] = src_snapshots_and_bookmarks.splitlines() 

240 src_snapshots_and_bookmarks = None 

241 if len(dst_snapshots_with_guids) == 0 and "bookmark" in types: 

242 # src bookmarks serve no purpose if the destination dataset has no snapshot; ignore them 

243 src_snapshots_with_guids = [snapshot for snapshot in src_snapshots_with_guids if "@" in snapshot] 

244 num_src_snapshots_found: int = sum(1 for snapshot in src_snapshots_with_guids if "@" in snapshot) 

245 with job.stats_lock: 

246 job.num_snapshots_found += num_src_snapshots_found 

247 # apply include/exclude regexes to ignore irrelevant src snapshots 

248 basis_src_snapshots_with_guids: list[str] = src_snapshots_with_guids 

249 src_snapshots_with_guids = filter_snapshots(job, src_snapshots_with_guids) 

250 if filter_needs_creation_time: 

251 src_snapshots_with_guids = cut(field=2, lines=src_snapshots_with_guids) 

252 basis_src_snapshots_with_guids = cut(field=2, lines=basis_src_snapshots_with_guids) 

253 

254 # find oldest and latest "true" snapshot, as well as GUIDs of all snapshots and bookmarks. 

255 # a snapshot is "true" if it is not a bookmark. 

256 oldest_src_snapshot: str = "" 

257 latest_src_snapshot: str = "" 

258 included_src_guids: set[str] = set() 

259 for line in src_snapshots_with_guids: 

260 guid, snapshot = line.split("\t", 1) 

261 if "@" in snapshot: 

262 included_src_guids.add(guid) 

263 latest_src_snapshot = snapshot 

264 if not oldest_src_snapshot: 

265 oldest_src_snapshot = snapshot 

266 if len(src_snapshots_with_guids) == 0: 

267 if p.skip_missing_snapshots == "fail": 

268 die(f"Source dataset includes no snapshot: {src_dataset}. Consider using --skip-missing-snapshots=dataset") 

269 elif p.skip_missing_snapshots == "dataset": 

270 log.warning("Skipping source dataset because it includes no snapshot: %s", src_dataset) 

271 if p.recursive and not job.dst_dataset_exists[dst_dataset]: 

272 log.warning("Also skipping descendant datasets as dst dataset does not exist for %s", src_dataset) 

273 return job.dst_dataset_exists[dst_dataset] 

274 return ( 

275 basis_src_snapshots_with_guids, 

276 src_snapshots_with_guids, 

277 dst_snapshots_with_guids, 

278 included_src_guids, 

279 latest_src_snapshot, 

280 oldest_src_snapshot, 

281 ) 

282 

283 

284def _rollback_dst_dataset_if_necessary( 

285 job: Job, 

286 dst_dataset: str, 

287 latest_src_snapshot: str, 

288 src_snapshots_with_guids: list[str], 

289 dst_snapshots_with_guids: list[str], 

290 done_checking: bool, 

291 tid: str, 

292) -> bool | tuple[str, str, bool]: 

293 """On replication, rollback dst if necessary; error out if not permitted.""" 

294 p, log = job.params, job.params.log 

295 dst = p.dst 

296 latest_dst_snapshot: str = "" 

297 latest_dst_guid: str = "" 

298 if len(dst_snapshots_with_guids) > 0: 

299 latest_dst_guid, latest_dst_snapshot = dst_snapshots_with_guids[-1].split("\t", 1) 

300 if p.force_rollback_to_latest_snapshot: 

301 log.info(p.dry(f"{tid} Rolling back destination to most recent snapshot: %s"), latest_dst_snapshot) 

302 # rollback just in case the dst dataset was modified since its most recent snapshot 

303 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset) 

304 cmd: list[str] = p.split_args(f"{dst.sudo} {p.zfs_program} rollback", latest_dst_snapshot) 

305 job.try_ssh_command(dst, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd, exists=False) 

306 elif latest_src_snapshot == "": 

307 log.info(f"{tid} Already-up-to-date: %s", dst_dataset) 

308 return True 

309 

310 # find most recent snapshot (or bookmark) that src and dst have in common - we'll start to replicate 

311 # from there up to the most recent src snapshot. any two snapshots are "common" iff their ZFS GUIDs (i.e. 

312 # contents) are equal. See https://github.com/openzfs/zfs/commit/305bc4b370b20de81eaf10a1cf724374258b74d1 

313 def latest_common_snapshot(snapshots_with_guids: list[str], intersect_guids: set[str]) -> tuple[str | None, str]: 

314 """Returns a true snapshot instead of its bookmark with the same GUID, per the sort order previously used for 'zfs 

315 list -s ...'.""" 

316 for _line in reversed(snapshots_with_guids): 

317 guid_, snapshot_ = _line.split("\t", 1) 

318 if guid_ in intersect_guids: 

319 return guid_, snapshot_ # can be a snapshot or bookmark 

320 return None, "" 

321 

322 latest_common_guid, latest_common_src_snapshot = latest_common_snapshot( 

323 src_snapshots_with_guids, set(cut(field=1, lines=dst_snapshots_with_guids)) 

324 ) 

325 log.debug("latest_common_src_snapshot: %s", latest_common_src_snapshot) # is a snapshot or bookmark 

326 log.log(LOG_TRACE, "latest_dst_snapshot: %s", latest_dst_snapshot) 

327 

328 if latest_common_src_snapshot and latest_common_guid != latest_dst_guid: 

329 # found latest common snapshot but dst has an even newer snapshot. rollback dst to that common snapshot. 

330 assert latest_common_guid 

331 _, latest_common_dst_snapshot = latest_common_snapshot(dst_snapshots_with_guids, {latest_common_guid}) 

332 if not (p.force_rollback_to_latest_common_snapshot or p.force): 

333 die( 

334 f"Conflict: Most recent destination snapshot {latest_dst_snapshot} is more recent than " 

335 f"most recent common snapshot {latest_common_dst_snapshot}. Rollback destination first, " 

336 "for example via --force-rollback-to-latest-common-snapshot (or --force) option." 

337 ) 

338 if p.force_once: 

339 p.force.value = False 

340 p.force_rollback_to_latest_common_snapshot.value = False 

341 log.info(p.dry(f"{tid} Rolling back destination to most recent common snapshot: %s"), latest_common_dst_snapshot) 

342 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset) 

343 cmd = p.split_args( 

344 f"{dst.sudo} {p.zfs_program} rollback -r {p.force_unmount} {p.force_hard}", latest_common_dst_snapshot 

345 ) 

346 try: 

347 job.run_ssh_command(dst, LOG_DEBUG, is_dry=p.dry_run, print_stdout=True, cmd=cmd) 

348 except (subprocess.CalledProcessError, UnicodeDecodeError) as e: 

349 stderr: str = stderr_to_str(e.stderr) if hasattr(e, "stderr") else "" 

350 retry_immediately_once: bool = _clear_resumable_recv_state_if_necessary(job, dst_dataset, stderr) 

351 # op isn't idempotent so retries regather current state from the start of replicate_dataset() 

352 raise RetryableError(display_msg="zfs rollback", retry_immediately_once=retry_immediately_once) from e 

353 

354 if latest_src_snapshot and latest_src_snapshot == latest_common_src_snapshot: 

355 log.info(f"{tid} Already up-to-date: %s", dst_dataset) 

356 return True 

357 return latest_dst_snapshot, latest_common_src_snapshot, done_checking 

358 

359 

360def _replicate_dataset_fully( 

361 job: Job, 

362 src_dataset: str, 

363 dst_dataset: str, 

364 oldest_src_snapshot: str, 

365 latest_src_snapshot: str, 

366 latest_dst_snapshot: str, 

367 dst_snapshots_with_guids: list[str], 

368 props_cache: dict[tuple[str, str, str], dict[str, str | None]], 

369 dry_run_no_send: bool, 

370 done_checking: bool, 

371 retry_count: int, 

372 tid: str, 

373) -> tuple[str, bool, bool, int]: 

374 """On replication, deletes all dst snapshots and performs a full send of the oldest selected src snapshot, which in turn 

375 creates a common snapshot; error out if not permitted.""" 

376 p, log = job.params, job.params.log 

377 src, dst = p.src, p.dst 

378 latest_common_src_snapshot: str = "" 

379 if latest_dst_snapshot: 

380 if not p.force: 

381 die( 

382 f"Conflict: No common snapshot found between {src_dataset} and {dst_dataset} even though " 

383 "destination has at least one snapshot. Aborting. Consider using --force option to first " 

384 "delete all existing destination snapshots in order to be able to proceed with replication." 

385 ) 

386 if p.force_once: 386 ↛ 388line 386 didn't jump to line 388 because the condition on line 386 was always true

387 p.force.value = False 

388 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset) 

389 # extract SNAPSHOT_TAG from GUID<TAB>DATASET@SNAPSHOT_TAG 

390 delete_snapshots(job, dst, dst_dataset, snapshot_tags=cut(2, separator="@", lines=dst_snapshots_with_guids)) 

391 if p.dry_run: 

392 # As we're in --dryrun (--force) mode this conflict resolution step (see above) wasn't really executed: 

393 # "no common snapshot was found. delete all dst snapshots". In turn, this would cause the subsequent 

394 # 'zfs receive -n' to fail with "cannot receive new filesystem stream: destination has snapshots; must 

395 # destroy them to overwrite it". So we skip the zfs send/receive step and keep on trucking. 

396 dry_run_no_send = True 

397 

398 # to start with, fully replicate oldest snapshot, which in turn creates a common snapshot 

399 if p.no_stream: 

400 oldest_src_snapshot = latest_src_snapshot 

401 if oldest_src_snapshot: 

402 if not job.dst_dataset_exists[dst_dataset]: 

403 # on destination, create parent filesystem and ancestors if they do not yet exist 

404 dst_dataset_parent: str = os.path.dirname(dst_dataset) 

405 if not job.dst_dataset_exists[dst_dataset_parent]: 

406 if p.dry_run: 

407 dry_run_no_send = True 

408 if dst_dataset_parent: 408 ↛ 411line 408 didn't jump to line 411 because the condition on line 408 was always true

409 _create_zfs_filesystem(job, dst_dataset_parent) 

410 

411 recv_resume_token_result: tuple[str | None, list[str], list[str]] = _recv_resume_token(job, dst_dataset, retry_count) 

412 recv_resume_token, send_resume_opts, recv_resume_opts = recv_resume_token_result 

413 curr_size: int = _estimate_send_size(job, src, dst_dataset, recv_resume_token, oldest_src_snapshot) 

414 humansize: str = _format_size(curr_size) 

415 if recv_resume_token: 

416 send_opts: list[str] = p.curr_zfs_send_program_opts + send_resume_opts # e.g. curr + ["-t", "1-c740b4779-..."] 

417 else: 

418 send_opts = p.curr_zfs_send_program_opts + [oldest_src_snapshot] 

419 send_cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} send", send_opts) 

420 recv_opts: list[str] = p.zfs_full_recv_opts.copy() + recv_resume_opts 

421 recv_opts, set_opts = _add_recv_property_options(job, True, recv_opts, src_dataset, props_cache) 

422 recv_cmd: list[str] = p.split_args( 

423 f"{dst.sudo} {p.zfs_program} receive -F", p.dry_run_recv, recv_opts, dst_dataset, allow_all=True 

424 ) 

425 log.info(p.dry(f"{tid} Full send: %s"), f"{oldest_src_snapshot} --> {dst_dataset} ({humansize.strip()}) ...") 

426 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset) 

427 dry_run_no_send = dry_run_no_send or p.dry_run_no_send 

428 job.maybe_inject_params(error_trigger="full_zfs_send_params") 

429 humansize = humansize.rjust(_RIGHT_JUST * 3 + 2) 

430 _run_zfs_send_receive( # do the real work 

431 job, src_dataset, dst_dataset, send_cmd, recv_cmd, curr_size, humansize, dry_run_no_send, "full_zfs_send" 

432 ) 

433 latest_common_src_snapshot = oldest_src_snapshot # we have now created a common snapshot 

434 if not dry_run_no_send and not p.dry_run: 

435 job.dst_dataset_exists[dst_dataset] = True 

436 with job.stats_lock: 

437 job.num_snapshots_replicated += 1 

438 _create_zfs_bookmarks(job, src, src_dataset, [oldest_src_snapshot]) 

439 _zfs_set(job, set_opts, dst, dst_dataset) 

440 dry_run_no_send = dry_run_no_send or p.dry_run 

441 retry_count = 0 

442 

443 return latest_common_src_snapshot, dry_run_no_send, done_checking, retry_count 

444 

445 

446def _replicate_dataset_incrementally( 

447 job: Job, 

448 src_dataset: str, 

449 dst_dataset: str, 

450 latest_common_src_snapshot: str, 

451 latest_src_snapshot: str, 

452 basis_src_snapshots_with_guids: list[str], 

453 included_src_guids: set[str], 

454 props_cache: dict[tuple[str, str, str], dict[str, str | None]], 

455 dry_run_no_send: bool, 

456 done_checking: bool, 

457 retry_count: int, 

458 tid: str, 

459) -> None: 

460 """Incrementally replicates all selected snapshots from latest common snapshot until latest src snapshot.""" 

461 p, log = job.params, job.params.log 

462 src, dst = p.src, p.dst 

463 

464 def replication_candidates() -> tuple[list[str], list[str]]: 

465 assert len(basis_src_snapshots_with_guids) > 0 

466 result_snapshots: list[str] = [] 

467 result_guids: list[str] = [] 

468 last_appended_guid: str = "" 

469 snapshot_itr: Iterator[str] = reversed(basis_src_snapshots_with_guids) 

470 while True: 

471 guid, snapshot = next(snapshot_itr).split("\t", 1) 

472 if "@" in snapshot: 

473 result_snapshots.append(snapshot) 

474 result_guids.append(guid) 

475 last_appended_guid = guid 

476 if snapshot == latest_common_src_snapshot: # latest_common_src_snapshot is a snapshot or bookmark 

477 if guid != last_appended_guid and "@" not in snapshot: 

478 # only appends the src bookmark if it has no snapshot. If the bookmark has a snapshot then that 

479 # snapshot has already been appended, per the sort order previously used for 'zfs list -s ...' 

480 result_snapshots.append(snapshot) 

481 result_guids.append(guid) 

482 break 

483 result_snapshots.reverse() 

484 result_guids.reverse() 

485 assert len(result_snapshots) > 0 

486 assert len(result_snapshots) == len(result_guids) 

487 return result_guids, result_snapshots 

488 

489 # collect the most recent common snapshot (which may be a bookmark) followed by all src snapshots 

490 # (that are not a bookmark) that are more recent than that. 

491 cand_guids, cand_snapshots = replication_candidates() 

492 if len(cand_snapshots) == 1: 

493 # latest_src_snapshot is a (true) snapshot that is equal to latest_common_src_snapshot or LESS recent 

494 # than latest_common_src_snapshot. The latter case can happen if latest_common_src_snapshot is a 

495 # bookmark whose snapshot has been deleted on src. 

496 return # nothing more tbd 

497 

498 recv_resume_token_result: tuple[str | None, list[str], list[str]] = _recv_resume_token(job, dst_dataset, retry_count) 

499 recv_resume_token, send_resume_opts, recv_resume_opts = recv_resume_token_result 

500 recv_opts: list[str] = p.zfs_recv_program_opts.copy() + recv_resume_opts 

501 recv_opts, set_opts = _add_recv_property_options(job, False, recv_opts, src_dataset, props_cache) 

502 if p.no_stream: 

503 # skip intermediate snapshots 

504 steps_todo: list[tuple[str, str, str, list[str]]] = [ 

505 ("-i", latest_common_src_snapshot, latest_src_snapshot, [latest_src_snapshot]) 

506 ] 

507 else: 

508 # include intermediate src snapshots that pass --{include,exclude}-snapshot-* policy, using 

509 # a series of -i/-I send/receive steps that skip excluded src snapshots. 

510 steps_todo = _incremental_send_steps_wrapper( 

511 p, cand_snapshots, cand_guids, included_src_guids, recv_resume_token is not None 

512 ) 

513 log.log(LOG_TRACE, "steps_todo: %s", list_formatter(steps_todo, "; ")) 

514 estimate_send_sizes: list[int] = _estimate_send_sizes_in_parallel(job, src, dst_dataset, recv_resume_token, steps_todo) 

515 total_size: int = sum(estimate_send_sizes) 

516 total_num: int = sum(len(to_snapshots) for incr_flag, from_snap, to_snap, to_snapshots in steps_todo) 

517 done_size: int = 0 

518 done_num: int = 0 

519 for i, (incr_flag, from_snap, to_snap, to_snapshots) in enumerate(steps_todo): 

520 curr_num_snapshots: int = len(to_snapshots) 

521 curr_size: int = estimate_send_sizes[i] 

522 humansize: str = _format_size(total_size) + "/" + _format_size(done_size) + "/" + _format_size(curr_size) 

523 human_num: str = f"{total_num}/{done_num}/{curr_num_snapshots} snapshots" 

524 if recv_resume_token: 

525 send_opts: list[str] = p.curr_zfs_send_program_opts + send_resume_opts # e.g. curr + ["-t", "1-c740b4779-..."] 

526 else: 

527 send_opts = p.curr_zfs_send_program_opts + [incr_flag, from_snap, to_snap] 

528 send_cmd: list[str] = p.split_args(f"{src.sudo} {p.zfs_program} send", send_opts) 

529 recv_cmd: list[str] = p.split_args( 

530 f"{dst.sudo} {p.zfs_program} receive", p.dry_run_recv, recv_opts, dst_dataset, allow_all=True 

531 ) 

532 dense_size: str = p.two_or_more_spaces_regex.sub("", humansize.strip()) 

533 log.info( 

534 p.dry(f"{tid} Incremental send {incr_flag}: %s"), 

535 f"{from_snap} .. {to_snap[to_snap.index('@'):]} --> {dst_dataset} ({dense_size}) ({human_num}) ...", 

536 ) 

537 done_checking = done_checking or _check_zfs_dataset_busy(job, dst, dst_dataset, busy_if_send=False) 

538 if p.dry_run and not job.dst_dataset_exists[dst_dataset]: 

539 dry_run_no_send = True 

540 dry_run_no_send = dry_run_no_send or p.dry_run_no_send 

541 job.maybe_inject_params(error_trigger="incr_zfs_send_params") 

542 _run_zfs_send_receive( # do the real work 

543 job, src_dataset, dst_dataset, send_cmd, recv_cmd, curr_size, humansize, dry_run_no_send, "incr_zfs_send" 

544 ) 

545 done_size += curr_size 

546 done_num += curr_num_snapshots 

547 recv_resume_token = None 

548 with job.stats_lock: 

549 job.num_snapshots_replicated += curr_num_snapshots 

550 assert p.create_bookmarks 

551 if p.create_bookmarks == "all": 

552 _create_zfs_bookmarks(job, src, src_dataset, to_snapshots) 

553 elif p.create_bookmarks != "none": 

554 threshold_millis: int = p.xperiods.label_milliseconds("_" + p.create_bookmarks) 

555 to_snapshots = [snap for snap in to_snapshots if p.xperiods.label_milliseconds(snap) >= threshold_millis] 

556 if i == len(steps_todo) - 1 and (len(to_snapshots) == 0 or to_snapshots[-1] != to_snap): 

557 to_snapshots.append(to_snap) # ensure latest common snapshot is bookmarked 

558 _create_zfs_bookmarks(job, src, src_dataset, to_snapshots) 

559 _zfs_set(job, set_opts, dst, dst_dataset) 

560 

561 

562def _format_size(num_bytes: int) -> str: 

563 """Formats a byte count for human-readable logs.""" 

564 return human_readable_bytes(num_bytes, separator="").rjust(_RIGHT_JUST) 

565 

566 

567def _prepare_zfs_send_receive( 

568 job: Job, src_dataset: str, send_cmd: list[str], recv_cmd: list[str], size_estimate_bytes: int, size_estimate_human: str 

569) -> tuple[str, str, str]: 

570 """Constructs zfs send/recv pipelines with optional compression, mbuffer and pv.""" 

571 p = job.params 

572 src, dst = p.src, p.dst 

573 send_cmd_str: str = shlex.join(send_cmd) 

574 recv_cmd_str: str = shlex.join(recv_cmd) 

575 src_has_shell: bool = p.is_program_available("sh", "src") 

576 dst_has_shell: bool = p.is_program_available("sh", "dst") 

577 same_remote: bool = is_same_remote(src, dst) 

578 r2r_mode: str = p.r2r_mode 

579 assert r2r_mode in ("off", "pull", "push"), r2r_mode 

580 

581 if ( 

582 p.is_program_available("zstd", "src") 

583 and p.is_program_available("zstd", "dst") 

584 and src_has_shell 

585 and dst_has_shell 

586 and (r2r_mode == "off" or not same_remote) 

587 ): 

588 compress_cmd_: str = _compress_cmd(p, "src", size_estimate_bytes) 

589 decompress_cmd_: str = _decompress_cmd(p, "dst", size_estimate_bytes) 

590 else: # no compression is used if source and destination do not both support compression 

591 compress_cmd_, decompress_cmd_ = "cat", "cat" 

592 

593 recordsize: int = abs(job.src_properties[src_dataset].recordsize) 

594 src_buffer: str = "cat" 

595 if r2r_mode == "off" or (src_has_shell and not same_remote): 

596 src_buffer = _mbuffer_cmd(p, "src", size_estimate_bytes, recordsize) 

597 dst_buffer: str = "cat" 

598 if r2r_mode == "off" or (dst_has_shell and not same_remote): 

599 dst_buffer = _mbuffer_cmd(p, "dst", size_estimate_bytes, recordsize) 

600 local_buffer: str = "cat" 

601 if r2r_mode == "off": 

602 local_buffer = _mbuffer_cmd(p, "local", size_estimate_bytes, recordsize) 

603 

604 pv_src_cmd: str = "" 

605 pv_dst_cmd: str = "" 

606 pv_loc_cmd: str = "" 

607 if r2r_mode == "off": 

608 if not p.src.ssh_user_host: 

609 pv_src_cmd = _pv_cmd(job, "local", size_estimate_bytes, size_estimate_human) 

610 elif not p.dst.ssh_user_host: 

611 pv_dst_cmd = _pv_cmd(job, "local", size_estimate_bytes, size_estimate_human) 

612 elif compress_cmd_ == "cat": 

613 pv_loc_cmd = _pv_cmd(job, "local", size_estimate_bytes, size_estimate_human) # compression disabled 

614 else: 

615 # pull-push mode with compression enabled: reporting "percent complete" isn't straightforward because 

616 # localhost observes the compressed data instead of the uncompressed data, so we disable the progress bar. 

617 pv_loc_cmd = _pv_cmd(job, "local", size_estimate_bytes, size_estimate_human, disable_progress_bar=True) 

618 

619 # assemble pipeline running on source leg 

620 src_pipe: str = "" 

621 if job.inject_params.get("inject_src_pipe_fail", False): 

622 # for testing; initially forward some bytes and then fail 

623 src_pipe = f"{src_pipe} | dd bs=64 count=1 2>/dev/null && false" 

624 if job.inject_params.get("inject_src_pipe_garble", False): 

625 src_pipe = f"{src_pipe} | gzip -1 -c -n" # for testing; forward garbled bytes 

626 if pv_src_cmd and pv_src_cmd != "cat": 

627 src_pipe = f"{src_pipe} | {pv_src_cmd}" 

628 if compress_cmd_ != "cat": 

629 src_pipe = f"{src_pipe} | {compress_cmd_}" 

630 if src_buffer != "cat": 

631 src_pipe = f"{src_pipe} | {src_buffer}" 

632 if src_pipe.startswith(" |"): 

633 src_pipe = src_pipe[2:] # strip leading ' |' part 

634 if job.inject_params.get("inject_src_send_error", False): 

635 send_cmd_str = f"{send_cmd_str} --injectedGarbageParameter" # for testing; induce CLI parse error 

636 if src_pipe: 

637 src_pipe = f"{send_cmd_str} | {src_pipe}" 

638 if p.src.ssh_user_host: 

639 src_pipe = p.shell_program + " -c " + dquote(src_pipe) 

640 else: 

641 src_pipe = send_cmd_str 

642 

643 # assemble pipeline running on middle leg between source and destination. only enabled for pull-push mode 

644 local_pipe: str = "" 

645 if r2r_mode == "off": 

646 if local_buffer != "cat": 

647 local_pipe = f"{local_buffer}" 

648 if pv_loc_cmd and pv_loc_cmd != "cat": 

649 local_pipe = f"{local_pipe} | {pv_loc_cmd}" 

650 if local_buffer != "cat": 

651 local_pipe = f"{local_pipe} | {local_buffer}" 

652 if local_pipe.startswith(" |"): 

653 local_pipe = local_pipe[2:] # strip leading ' |' part 

654 if local_pipe: 

655 local_pipe = f"| {local_pipe}" 

656 

657 # assemble pipeline running on destination leg 

658 dst_pipe: str = "" 

659 if dst_buffer != "cat": 

660 dst_pipe = f"{dst_buffer}" 

661 if decompress_cmd_ != "cat": 

662 dst_pipe = f"{dst_pipe} | {decompress_cmd_}" 

663 if pv_dst_cmd and pv_dst_cmd != "cat": 

664 dst_pipe = f"{dst_pipe} | {pv_dst_cmd}" 

665 if job.inject_params.get("inject_dst_pipe_fail", False): 

666 # interrupt zfs receive for testing retry/resume; initially forward some bytes and then stop forwarding 

667 dst_pipe = f"{dst_pipe} | dd bs=1024 count={INJECT_DST_PIPE_FAIL_KBYTES} 2>/dev/null" 

668 if job.inject_params.get("inject_dst_pipe_garble", False): 

669 dst_pipe = f"{dst_pipe} | gzip -1 -c -n" # for testing; forward garbled bytes 

670 if dst_pipe.startswith(" |"): 

671 dst_pipe = dst_pipe[2:] # strip leading ' |' part 

672 if job.inject_params.get("inject_dst_receive_error", False): 

673 recv_cmd_str = f"{recv_cmd_str} --injectedGarbageParameter" # for testing; induce CLI parse error 

674 if dst_pipe: 

675 dst_pipe = f"{dst_pipe} | {recv_cmd_str}" 

676 if p.dst.ssh_user_host: 

677 dst_pipe = p.shell_program + " -c " + dquote(dst_pipe) 

678 else: 

679 dst_pipe = recv_cmd_str 

680 

681 if r2r_mode == "off": 

682 # If there's no support for shell pipelines, we can't do compression, mbuffering, monitoring and rate-limiting, 

683 # so we fall back to simple zfs send/receive. 

684 if not src_has_shell: 

685 src_pipe = send_cmd_str 

686 if not dst_has_shell: 

687 dst_pipe = recv_cmd_str 

688 

689 src_pipe = squote(p.src, src_pipe) 

690 dst_pipe = squote(p.dst, dst_pipe) 

691 return src_pipe, local_pipe, dst_pipe 

692 

693 def nested_ssh_cmd(remote: Remote) -> str: 

694 """Builds nested ssh command text for r2r leg.""" 

695 if same_remote: 

696 # Perf: use local replication if initiator sees src+dst have same user@host, port and ssh configfile. 

697 # Example: ssh alice@src.example.com "sh -c 'zfs send <src_dataset> | zfs receive <dst_dataset>'" 

698 return "" 

699 cmd: list[str] = [p.ssh_program] + list(remote.ssh_extra_opts) 

700 if remote.ssh_config_file: 

701 cmd += ["-F", remote.ssh_config_file] 

702 if remote.ssh_cipher: 

703 cmd += ["-c", remote.ssh_cipher] 

704 if remote.ssh_port: 

705 cmd += ["-p", str(remote.ssh_port)] 

706 assert remote.ssh_user_host 

707 cmd.append(remote.ssh_user_host) 

708 return shlex.join(cmd) 

709 

710 if r2r_mode == "push": 

711 # Example: ssh alice@src.example.com "sh -c 'zfs send ... | ssh bob@dst.example.com zfs receive ...'" 

712 dst_pipe = dst_pipe if same_remote else shlex.quote(dst_pipe) 

713 r2r_cmd = f"{src_pipe} | {nested_ssh_cmd(dst)} {dst_pipe}" 

714 else: 

715 assert r2r_mode == "pull", r2r_mode 

716 # Example: ssh bob@dst.example.com "sh -c 'ssh alice@src.example.com zfs send ... | zfs receive ...'" 

717 src_pipe = src_pipe if same_remote else shlex.quote(src_pipe) 

718 r2r_cmd = f"{nested_ssh_cmd(src)} {src_pipe} | {dst_pipe}" 

719 r2r_cmd = shlex.quote(p.shell_program + " -c " + shlex.quote(r2r_cmd)) 

720 return r2r_cmd, "", "" 

721 

722