bscp.python2

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# Copyright (C) 2012-2024  The Bscp Authors <https://github.com/bscp-tool/bscp/graphs/contributors>
#
# This is a fork.  It builds on, and is inspired by, the original bscp:
#   https://github.com/bscp-tool/bscp   (canonical project, latest master)
#   https://github.com/vog/bscp         (original author, Volker Diels-Grabsch)
# Fork: https://github.com/groenewe/bscp
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

# Bscp copies a file or block device over SSH, transferring only changed
# blocks.  Both push (local → remote) and pull (remote → local) are
# supported.  The remote-side script is embedded and executed via
# "python(/2/3) -c", with a Perl fallback for hosts without Python, so
# no installation on the remote host is required.
# Features: section-based processing to bound memory use, single-pass
# scan+copy, windowed pull to avoid SSH pipe deadlock, resume on failure
# (byte offset or percentage of source), automatic retry with exponential
# back-off, and dry-run mode.
#
# Changes in this fork were made with extensive help of Claude Code, Sonnet
# and Opus models (Anthropic).

from __future__ import division

from argparse import ArgumentParser, ArgumentTypeError
import binascii
import hashlib
import math
import os
import shlex
import struct
import subprocess
import sys
from time import sleep, time

# Python 2.7 has shlex but no shlex.quote; the pipes module provides the same
# function under the same name.  Python 3.3+ has shlex.quote in stdlib.
try:
    from shlex import quote as _shquote
except ImportError:
    from pipes import quote as _shquote

# BrokenPipeError / ConnectionResetError are Python 3.3+; on Python 2 a broken
# pipe surfaces as IOError(EPIPE) and a reset connection as a generic
# IOError/OSError.  Alias them to the Py2 equivalents so the same except
# clauses work verbatim under both interpreters: a literal `except
# BrokenPipeError` would raise NameError on Py2 *while* handling a real pipe
# error and mask it.  The named aliases let each handler list exactly the
# exceptions it needs (some want OSError, some EOFError, some neither) instead
# of forcing one shared tuple everywhere.
try:
    _BrokenPipeError      = BrokenPipeError
    _ConnectionResetError = ConnectionResetError
except NameError:
    _BrokenPipeError      = IOError
    _ConnectionResetError = OSError
_PIPE_ERRORS = (_BrokenPipeError, _ConnectionResetError, EOFError)

# Header: size(8), blocksize(8), section_size(8), start_offset(8), filename_len(8), hashname_len(8), mode(1) - 49 bytes total
HEADER_FMT  = '<QQQQQQB'
HEADER_SIZE = struct.calcsize(HEADER_FMT)
MODE_PUSH, MODE_PULL = 0, 1
ALLOW_TRUNCATE = 2

# Number of diff-block requests sent per round-trip in pull phase B.
# Larger = fewer round-trips; too large risks filling SSH stdout window before
# the client drains it.  128 × 64 KiB = 8 MiB max in-flight, well within
# OpenSSH's dynamic window.
PULL_WINDOW = 128

DEFAULT_BLOCK   = 64 * 1024
DEFAULT_SECTION = 10 * 1024**3
DEFAULT_RETRIES = 3
MIN_MEMORY_PUSH = 64 * DEFAULT_BLOCK

# Hash algorithms guaranteed on every remote variant: python3, python2, and
# the Perl fallback (Digest::MD5 + Digest::SHA).  hashlib offers more
# locally, but only these are portable to all three remotes.
PORTABLE_ALGOS = ('md5', 'sha1', 'sha224', 'sha256', 'sha384', 'sha512')


def algorithm_help():
    # Local-only extras: anything hashlib can do here that is not in the
    # portable set.  Drop uppercase aliases (e.g. 'SHA256') for a clean list,
    # and skip XOF/variable-length functions (shake_*, digest_size 0) — they
    # have no fixed digest and would break the wire protocol.
    portable = set(PORTABLE_ALGOS)
    extra = sorted(a for a in hashlib.algorithms_available
                   if a.lower() not in portable and a == a.lower()
                   and hashlib.new(a).digest_size > 0)
    h = ('hash algorithm (default sha256). Portable to all remotes: %s.'
         % ', '.join(PORTABLE_ALGOS))
    if extra:
        h += (' Also usable locally: %s — these need the remote to be '
              'python3/python2 with the same algorithm; the Perl remote '
              'supports only the portable set.' % ', '.join(extra))
    return h


def parse_algorithm(name):
    # Reject anything the local hashlib cannot build, and XOF/variable-length
    # functions (shake_*, digest_size 0) — the wire protocol assumes a fixed
    # digest size per block.  Remote availability still varies (Perl supports
    # only PORTABLE_ALGOS); that surfaces as a remote handshake error.
    try:
        if hashlib.new(name).digest_size <= 0:
            raise ArgumentTypeError(
                "hash algorithm %r has no fixed digest size and cannot be "
                "used (variable-length/XOF functions are unsupported)" % name)
    except ArgumentTypeError:
        raise
    except (ValueError, TypeError):
        raise ArgumentTypeError("unknown hash algorithm %r" % name)
    return name

REPORT_INTERVAL = 0.5 # status line should update after this many seconds have passed
last_report_len = 0   # last progress or error length message on screen, if any; used by report()
quiet = False         # option -q or --quiet: suppress '\r' progress report line normally updated via stderr
batch = False         # option --batch: suppress all stderr output, return only a suitable exit code (see help)

# NOTE: the three remote source literals below (remote_script,
# remote_script_mt, remote_perl) are indented with TABS, not spaces.  Two of
# them are hex-encoded onto the ssh command line; a tab is one character
# where the equivalent 4-space indent is four, so tabs roughly halve the
# leading-whitespace bytes on the wire (deep nesting shrinks the most).  Keep
# them tab-indented when editing.  The surrounding bscp client code stays
# 4-space indented as usual — the tabs live only inside these string
# literals, so they do not affect this file's own indentation.  A `ts=4`
# vim modeline at the end of the file renders the embedded tabs at width 4.
remote_script = '''
def _remote():
	import hashlib, os, struct, sys  # noqa: F401 (standalone on remote host)

	stdin  = getattr(sys.stdin,  'buffer', sys.stdin)
	stdout = getattr(sys.stdout, 'buffer', sys.stdout)

	# Read exactly n bytes or raise EOFError.  On a blocking pipe a short
	# read means EOF, i.e. the client's SSH connection is gone; signal it
	# rather than acting on a truncated read (a partial block write would
	# otherwise corrupt the destination silently).
	def rd(n):
		buf = b''
		while len(buf) < n:
			chunk = stdin.read(n - len(buf))
			if not chunk:
				raise EOFError
			buf += chunk
		return buf

	HEADER_FMT     = '<QQQQQQB'
	HEADER_SIZE    = struct.calcsize(HEADER_FMT)
	MODE_PUSH, MODE_PULL = 0, 1
	PUSH_PULL_MASK = 1
	ALLOW_TRUNCATE = 2

	try:
		data = rd(HEADER_SIZE)
	except EOFError:
		sys.exit(1)
	(size, blocksize, section_size, start_offset,
	 filename_len, hashname_len, mode) = struct.unpack(HEADER_FMT, data)
	allow_truncate = mode & ALLOW_TRUNCATE
	mode = mode & PUSH_PULL_MASK

	try:
		fn_bytes = rd(filename_len)
		hashname = rd(hashname_len).decode('ascii')
		filename = fn_bytes.decode('utf-8')

		stdout.write(hashlib.new(hashname, fn_bytes).digest())
		stdout.flush()
		if rd(2) != b'go':
			sys.exit(1)
	except EOFError:
		sys.exit(1)

	try:
		fobj = open(filename, 'rb+' if mode == MODE_PUSH else 'rb')
	except OSError:
		stdout.write(struct.pack('<Q', 0))
		stdout.flush()
		sys.exit(1)

	with fobj as f:
		f.seek(0, 2)
		remote_size = f.tell()
		stdout.write(struct.pack('<Q', remote_size))
		stdout.flush()

		# In push, remote is destination; in pull, remote is source.
		# Refuse to proceed if destination is smaller, unless --allow-truncate.
		dst_size = remote_size if mode == MODE_PUSH else size
		src_size = size        if mode == MODE_PUSH else remote_size
		if dst_size < src_size and not allow_truncate:
			sys.exit(1)

		sync_size = min(size, remote_size)
		eff_section = section_size or sync_size
		if sync_size == 0:
			sys.exit(0)

		# If the client decides to abort after the handshake (e.g. its
		# post-handshake truncate check fires) it closes its end of the
		# pipe, which surfaces here as a short stdin read or a broken stdout
		# write.  Treat both as a clean shutdown rather than letting Python
		# spew a traceback over the user's stderr.
		try:
			pos = start_offset
			while pos < sync_size:
				sec_end = min(pos + eff_section, sync_size)

				p = pos
				while p < sec_end:
					bl = min(blocksize, sec_end - p)
					f.seek(p)
					stdout.write(hashlib.new(hashname, f.read(bl)).digest())
					p += bl
				stdout.flush()

				(count,) = struct.unpack('<Q', rd(8))

				if mode == MODE_PUSH:
					for _ in range(count):
						(wpos,) = struct.unpack('<Q', rd(8))
						bl = min(blocksize, sync_size - wpos)
						f.seek(wpos)
						f.write(rd(bl))
				else:
					for _ in range(count):
						(rpos,) = struct.unpack('<Q', rd(8))
						bl = min(blocksize, sync_size - rpos)
						f.seek(rpos)
						stdout.write(f.read(bl))
						stdout.flush()

				pos = sec_end
		except (struct.error, IOError, OSError, EOFError):
			sys.exit(1)
'''

# Multi-threaded python3 remote.  Functionally identical to remote_script
# on the wire — same handshake, same phase A/B contract — but phase-A
# hashing fans out across a ThreadPoolExecutor.  python3-only (uses
# concurrent.futures, sys.stdin.buffer unconditionally), so build_ssh_cmd()
# dispatches it ONLY to python3 and falls back to the single-threaded
# remote_script for python2/python and to remote_perl for Perl-only hosts.
#
# Because build_ssh_cmd() hex-encodes this and runs it via
#   python3 -O -B -c "exec(bytes.fromhex('...').decode())"
# the no-"/no-$/no-% constraints that apply to remote_script do NOT apply
# here — the hex alphabet is inert in every shell quoting context.  The
# thread count is baked into the trailing _remote(N) call by build_ssh_cmd
# (N = client --hash-threads; 0 = let the remote auto-detect its own cores).
#
# Protocol constants are duplicated here too (see CLAUDE.md): any change to
# HEADER_FMT / the mode bits / the phase contract must be made in FOUR
# places now — client, remote_script, remote_script_mt, and remote_perl.
remote_script_mt = '''
def _remote(nthreads):
	import hashlib, os, struct, sys
	from collections import deque
	from concurrent.futures import ThreadPoolExecutor

	stdin  = sys.stdin.buffer
	stdout = sys.stdout.buffer

	def rd(n):
		buf = b""
		while len(buf) < n:
			chunk = stdin.read(n - len(buf))
			if not chunk:
				raise EOFError
			buf += chunk
		return buf

	HEADER_FMT     = "<QQQQQQB"
	HEADER_SIZE    = struct.calcsize(HEADER_FMT)
	MODE_PUSH, MODE_PULL = 0, 1
	PUSH_PULL_MASK = 1
	ALLOW_TRUNCATE = 2
	HASH_THREADS_CAP = 4

	if nthreads <= 0:
		nthreads = min(os.cpu_count() or 1, HASH_THREADS_CAP)
	window = max(2, nthreads * 2)

	try:
		data = rd(HEADER_SIZE)
	except EOFError:
		sys.exit(1)
	(size, blocksize, section_size, start_offset,
	 filename_len, hashname_len, mode) = struct.unpack(HEADER_FMT, data)
	allow_truncate = mode & ALLOW_TRUNCATE
	mode = mode & PUSH_PULL_MASK

	try:
		fn_bytes = rd(filename_len)
		hashname = rd(hashname_len).decode("ascii")
		filename = fn_bytes.decode("utf-8")
		stdout.write(hashlib.new(hashname, fn_bytes).digest())
		stdout.flush()
		if rd(2) != b"go":
			sys.exit(1)
	except EOFError:
		sys.exit(1)

	try:
		fobj = open(filename, "rb+" if mode == MODE_PUSH else "rb")
	except OSError:
		stdout.write(struct.pack("<Q", 0))
		stdout.flush()
		sys.exit(1)

	def hashfn(block):
		return hashlib.new(hashname, block).digest()

	with fobj as f:
		f.seek(0, 2)
		remote_size = f.tell()
		stdout.write(struct.pack("<Q", remote_size))
		stdout.flush()

		dst_size = remote_size if mode == MODE_PUSH else size
		src_size = size        if mode == MODE_PUSH else remote_size
		if dst_size < src_size and not allow_truncate:
			sys.exit(1)

		sync_size = min(size, remote_size)
		eff_section = section_size or sync_size
		if sync_size == 0:
			sys.exit(0)

		ex = ThreadPoolExecutor(max_workers=nthreads)
		try:
			pos = start_offset
			while pos < sync_size:
				sec_end = min(pos + eff_section, sync_size)

				# Phase A: reads stay single-threaded and sequential (fast on
				# NVMe); only hashing fans out.  Digests are written in block
				# order, so the wire stream is identical to remote_script.
				f.seek(pos)
				p = pos
				inflight = deque()
				while p < sec_end and len(inflight) < window:
					bl = min(blocksize, sec_end - p)
					inflight.append(ex.submit(hashfn, f.read(bl)))
					p += bl
				while inflight:
					stdout.write(inflight.popleft().result())
					if p < sec_end:
						bl = min(blocksize, sec_end - p)
						inflight.append(ex.submit(hashfn, f.read(bl)))
						p += bl
				stdout.flush()

				(count,) = struct.unpack("<Q", rd(8))

				if mode == MODE_PUSH:
					for _ in range(count):
						(wpos,) = struct.unpack("<Q", rd(8))
						bl = min(blocksize, sync_size - wpos)
						f.seek(wpos)
						f.write(rd(bl))
				else:
					for _ in range(count):
						(rpos,) = struct.unpack("<Q", rd(8))
						bl = min(blocksize, sync_size - rpos)
						f.seek(rpos)
						stdout.write(f.read(bl))
						stdout.flush()

				pos = sec_end
		except (struct.error, IOError, OSError, EOFError):
			sys.exit(1)
		finally:
			ex.shutdown(wait=False)
'''

# Perl fallback for hosts that have no Python.  Functionally identical to
# remote_script above; see docs/remote-execution.md ("Remote script
# constraints") for why this exists as a hex-pack-eval payload rather than
# an inline -c argument.
remote_perl = '''
use strict; use warnings;
binmode STDIN; binmode STDOUT;

my $HEADER_FMT     = 'Q< Q< Q< Q< Q< Q< C';
my $HEADER_SIZE    = 49;
my ($MODE_PUSH, $MODE_PULL) = (0, 1);
my $PUSH_PULL_MASK = 1;
my $ALLOW_TRUNCATE = 2;

sub r {
	my $n = shift;
	my $buf = '';
	while (length($buf) < $n) {
		my $chunk;
		my $got = read(STDIN, $chunk, $n - length($buf));
		die "short read\\n" unless defined($got) && $got > 0;
		$buf .= $chunk;
	}
	return $buf;
}

sub make_hash {
	my $name = lc(shift);
	if ($name eq 'md5') {
		require Digest::MD5;
		return \\&Digest::MD5::md5;
	}
	if ($name =~ /^sha(1|224|256|384|512)\\z/) {
		require Digest::SHA;
		my $bits = $1;
		return sub { Digest::SHA->new($bits)->add($_[0])->digest };
	}
	die "unsupported hash algorithm: $name\\n";
}

eval {
	my $data = r($HEADER_SIZE);
	my ($size, $blocksize, $section_size, $start_offset,
		$filename_len, $hashname_len, $mode_byte) = unpack($HEADER_FMT, $data);
	my $allow_truncate = $mode_byte & $ALLOW_TRUNCATE;
	my $mode           = $mode_byte & $PUSH_PULL_MASK;

	my $fn_bytes = r($filename_len);
	my $hashname = r($hashname_len);

	my $hashfn = make_hash($hashname);
	print STDOUT $hashfn->($fn_bytes);
	STDOUT->flush;

	exit 1 unless r(2) eq 'go';

	my $fh;
	unless (open($fh, $mode == $MODE_PUSH ? '+<:raw' : '<:raw', $fn_bytes)) {
		print STDOUT pack('Q<', 0); STDOUT->flush; exit 1;
	}
	seek($fh, 0, 2);
	my $remote_size = tell($fh);
	print STDOUT pack('Q<', $remote_size); STDOUT->flush;

	my $dst_size = $mode == $MODE_PUSH ? $remote_size : $size;
	my $src_size = $mode == $MODE_PUSH ? $size        : $remote_size;
	exit 1 if $dst_size < $src_size && !$allow_truncate;

	my $sync_size   = $size < $remote_size ? $size : $remote_size;
	my $eff_section = $section_size || $sync_size;
	exit 0 if $sync_size == 0;

	my $pos = $start_offset;
	while ($pos < $sync_size) {
		my $sec_end = $pos + $eff_section;
		$sec_end = $sync_size if $sec_end > $sync_size;

		my $p = $pos;
		while ($p < $sec_end) {
			my $bl = $blocksize < $sec_end - $p ? $blocksize : $sec_end - $p;
			seek($fh, $p, 0);
			my $buf; read($fh, $buf, $bl);
			print STDOUT $hashfn->($buf);
			$p += $bl;
		}
		STDOUT->flush;

		my ($count) = unpack('Q<', r(8));

		if ($mode == $MODE_PUSH) {
			for (1..$count) {
				my ($wpos) = unpack('Q<', r(8));
				my $bl = $blocksize < $sync_size - $wpos ? $blocksize : $sync_size - $wpos;
				my $blk = r($bl);
				seek($fh, $wpos, 0);
				print $fh $blk;
			}
		} else {
			for (1..$count) {
				my ($rpos) = unpack('Q<', r(8));
				my $bl = $blocksize < $sync_size - $rpos ? $blocksize : $sync_size - $rpos;
				seek($fh, $rpos, 0);
				my $buf; read($fh, $buf, $bl);
				print STDOUT $buf;
				STDOUT->flush;
			}
		}
		$pos = $sec_end;
	}
};
exit 1 if $@;
exit 0;
'''


class IOCounter:
    def __init__(self, in_stream, out_stream, bwlimit=0):
        self.in_stream  = in_stream
        self.out_stream = out_stream
        self.in_total   = 0
        self.out_total  = 0
        # Combined-direction token bucket (mirrors the python3 client); 0 =
        # unlimited.  _bw_allow starts full so the first read/write is free.
        self.bwlimit    = bwlimit
        self._bw_allow  = float(bwlimit)
        self._bw_last   = time()

    def _throttle(self, nbytes):
        if not self.bwlimit:
            return
        now = time()
        # Refill, capped at one second's worth so an idle gap cannot bank
        # credit for a later flood.  Leave a negative balance negative: the
        # next call's refill is sized by the elapsed sleep and repays exactly
        # this deficit; zeroing it here would double-credit and run ~2x over.
        self._bw_allow = min(self._bw_allow + (now - self._bw_last) * self.bwlimit,
                             float(self.bwlimit))
        self._bw_last  = now
        self._bw_allow -= nbytes
        if self._bw_allow < 0:
            sleep(-self._bw_allow / self.bwlimit)

    def read(self, n):
        data = self.in_stream.read(n)
        self.in_total += len(data)
        self._throttle(len(data))
        return data

    def write(self, data):
        self.out_stream.write(data)
        self.out_total += len(data)
        self.out_stream.flush()
        self._throttle(len(data))


class ConnectionLost(RuntimeError):
    def __init__(self, offset, interrupted=False):
        verb = 'Interrupted' if interrupted else 'Connection lost'
        # super(Cls, self) form for Python 2 compatibility.
        super(ConnectionLost, self).__init__(
            '%s — resume with: --resume-from %d' % (verb, offset))
        self.offset      = offset
        self.interrupted = interrupted


def _shutdown_proc(proc):
    # close() flushes the BufferedWriter; against a dead pipe that flush
    # itself raises BrokenPipeError, which would chain on top of whatever
    # we were already handling and escape do_sync.  Same for stdout if
    # the peer reset between writes.  Swallow pipe errors here — by the
    # time we are tearing down, the wire is already gone.
    for stream in (proc.stdin, proc.stdout):
        try:
            stream.close()
        except (_BrokenPipeError, _ConnectionResetError, OSError):
            pass
    try:
        proc.wait()
    except OSError:
        pass


def report(msg):
    global last_report_len

    if batch:
        return
    if quiet and msg[:1] == '\r':
        return
    stt = 1 if (last_report_len == 0 and msg[:1] in ('\r', '\n')) else 0
    lines = msg[stt:].split('\n')
    # A leading '\r' repositions the cursor but occupies no column, so exclude
    # it from all length math.  Otherwise the pad and stored length disagree by
    # one between a first line that consumed its '\r' (stt=1) and a later one
    # that kept it (stt=0) — leaving one stale char (e.g. a trailing "0") when a
    # short final summary overwrites a single, longer scan/progress line.
    cr = '\r' if lines[0][:1] == '\r' else ''
    visible = lines[0][len(cr):]
    if last_report_len > len(visible):
        visible = visible + ' ' * (last_report_len - len(visible))
    lines[0] = cr + visible
    sys.stderr.write('\n'.join(lines))
    sys.stderr.flush()
    # last_report_len tracks chars left on the cursor's line so the next '\r'
    # overwrite can pad them away.  A msg ending in '\n' leaves the cursor on a
    # fresh empty line, so reset to 0 — otherwise the next call would pad
    # against this line's length and prepend stray spaces.
    last_report_len = 0 if msg.endswith('\n') else len(visible.rstrip())


def parse_size(s):
    s = s.strip()
    if not s:
        raise ValueError('empty size string')
    suffixes = {'k': 1024, 'm': 1024**2, 'g': 1024**3, 't': 1024**4}
    if s[-1].lower() in suffixes:
        return int(float(s[:-1]) * suffixes[s[-1].lower()])
    return int(s)


def parse_block_count(s):
    """Parse -B value. Bare integer N -> N blocks. With K/M/G/T suffix -> byte
    size, returned as a negative sentinel that main() resolves to a block
    count once the (possibly user-overridden) block size is known."""
    s = s.strip()
    if not s:
        raise ValueError('empty block-count value')
    if s[-1].lower() in ('k', 'm', 'g', 't'):
        v = parse_size(s)
        if v < 0:
            raise ValueError('block-count value must be non-negative')
        return -v if v else 0
    n = int(s)
    if n < 0:
        raise ValueError('block-count value must be non-negative')
    return n


def parse_resume_from(s):
    """Parse -r value. Bare integer or K/M/G/T size string -> bytes (positive).
    NN% or NN.N% (0..100 inclusive) -> negative permille sentinel that
    main() resolves to a byte offset against the local file size."""
    s = s.strip()
    if not s:
        raise ValueError('empty resume-from value')
    if s.endswith('%'):
        v = float(s[:-1])
        if not (0 <= v <= 100):
            raise ValueError('resume-from percentage must be between 0 and 100')
        return -int(round(v * 10))
    return parse_size(s)


def device_size(path):
    # os.path.getsize() reports 0 for block devices (stat st_size is 0); a
    # seek-to-end gives the true size for both regular files and block
    # devices — the same method do_sync and the remote use for their sizes.
    f = open(path, 'rb')
    try:
        f.seek(0, 2)
        return f.tell()
    finally:
        f.close()


def available_memory():
    try:
        pages = os.sysconf('SC_AVPHYS_PAGES')
        size  = os.sysconf('SC_PAGE_SIZE')
        if pages > 0 and size > 0:
            return pages * size
    except (AttributeError, ValueError, OSError):
        pass
    return 0


def format_time(seconds, round_eta=False):
    s = int(seconds)
    h, s = divmod(s, 3600)
    m, s = divmod(s, 60)
    if round_eta:
        if (h > 0 or m > 0):
            if s >= 10:
                m += 1
                if m >= 60:
                    m -= 60
                    h += 1
            s = 0
        elif (s >= 10):
            s = (s // 10) * 10
    return '%d:%02d:%02d' % (h, m, s) if h else '%d:%02d' % (m, s)


def format_size(n, floor=False):
    """Return n as a human-readable size string.
    floor=False (default): only emit a unit when n is an exact multiple of
        it.  Round-trips losslessly through parse_size(); use for CLI args.
    floor=True: truncate to the largest unit that keeps the count to four
        digits or fewer; use for display values.  When the next unit up
        would otherwise force a count of five or more digits, the count is
        allowed to drop below ten — e.g. 9999M displays as "9999M" but
        10000M (= 9.77G) becomes "9G", and 10240M (= 10G) becomes "10G"."""
    if floor:
        for suffix, div in (('',  1),
                            ('K', 1 << 10),
                            ('M', 1 << 20),
                            ('G', 1 << 30),
                            ('T', 1 << 40)):
            count = n // div
            if count <= 9999:
                return '%d%s' % (count, suffix)
        # n >= 10000 T (≈ 10 PiB) — fall back to T regardless of width.
        return '%dT' % (n >> 40)
    for suffix, div in (('T', 1 << 40), ('G', 1 << 30), ('M', 1 << 20), ('K', 1 << 10)):
        if n % div == 0 and n // div > 0:
            return '%d%s' % (n // div, suffix)
    return str(n)


def build_resume_cmd(argv0, args, src, dst, offset, block_count=None):
    parts = [argv0]
    if args.block_size != DEFAULT_BLOCK:
        parts += ['-b', format_size(args.block_size)]
    if args.section_size != DEFAULT_SECTION:
        parts += ['-s', format_size(args.section_size)]
    if args.algorithm != 'sha256':
        parts += ['-a', args.algorithm]
    if args.retries != DEFAULT_RETRIES:
        parts += ['--retries', str(args.retries)]
    if args.bwlimit:
        parts += ['--bwlimit', format_size(args.bwlimit)]
    if args.identity:
        parts += ['-i', args.identity]
    for opt in (args.ssh_opt or []):
        parts += ['-o', opt]
    if args.compress:
        parts.append('-C')
    if args.dry_run:
        parts.append('-N')
    if args.allow_truncate:
        parts.append('--allow-truncate')
    if args.buffer:
        parts.append('--buffer')
    bc = args.block_count if block_count is None else block_count
    if bc:
        parts += ['-B', str(bc)]
    if args.port != 22:
        parts += ['-p', str(args.port)]
    if args.quiet:
        parts.append('-q')
    if args.batch:
        parts.append('--batch')
    parts += ['-r', format_size(offset)]
    parts += [src, dst]
    return ' '.join(_shquote(p) for p in parts)


def build_ssh_cmd(remote_host, args):
    cmd = ['ssh']
    if args.get('compress'):
        cmd.append('-C')
    if args.get('port', 22) != 22:
        cmd += ['-p', str(args['port'])]
    if args.get('identity'):
        cmd += ['-i', args['identity']]
    for opt in (args.get('ssh_opt') or []):
        cmd += ['-o', opt]
    # Keepalive failsafe: a silently-dropped TCP connection (router NAT
    # timeout, peer kernel stall, ISP blip) otherwise leaves both ends
    # blocked on read/write forever.  With these settings ssh declares the
    # connection dead after ~60s of unresponsiveness, which surfaces as a
    # BrokenPipeError on the client and engages the ConnectionLost path
    # (and --retries, if set).  Appended after the user's -o options
    # because ssh applies the first matching -o, so a user-supplied
    # override still wins.
    cmd += ['-o', 'ServerAliveInterval=15', '-o', 'ServerAliveCountMax=4']
    # Each exec'd remote ends with a literal `bscp-remote` argument.  The
    # remote bodies never read argv, so it is inert — but it lands in the
    # process command line, so `ps aux | grep bscp-remote` (or htop search)
    # locates the remote process on the destination host.
    script_py = remote_script + '\n_remote()'
    # bytes.hex() is Py3.5+; binascii.hexlify works on both.  decode('ascii')
    # gives a native str on Py3 and unicode on Py2 — both interpolate fine
    # into the wrapper shell string below.
    hex_pl    = binascii.hexlify(remote_perl.encode('utf-8')).decode('ascii')
    # A python3 remote runs the multi-threaded remote_script_mt regardless of
    # which client drives it.  This py2 client has no --hash-threads option,
    # so the baked thread count is 0 (the remote auto-detects its own cores).
    # python2/python fall back to the single-threaded remote_script.  The
    # source is hex-encoded so the shell sees only inert hex inside the -c
    # arg, sidestepping the no-"/no-$/no-% rules.
    script_mt = remote_script_mt + '\n_remote(0)\n'
    hex_mt    = binascii.hexlify(script_mt.encode('utf-8')).decode('ascii')
    # BSCP_FORCE_PERL skips both Python branches; BSCP_FORCE_PYTHON2 skips the
    # python3/remote_script_mt branch so the single-threaded legacy
    # remote_script is exercised even where python3 is present.  Both are
    # test hooks (tests.sh) for paths that a fully-equipped host would
    # otherwise never reach.
    if os.environ.get('BSCP_FORCE_PERL'):
        py_branch = ''
    elif os.environ.get('BSCP_FORCE_PYTHON2'):
        py_branch = (
            'for py in python3 python2 python; do '
            'command -v "$py" >/dev/null 2>&1 && exec "$py" -O -B -c "%s" bscp-remote; '
            'done; ' % script_py)
    else:
        py_branch = (
            'command -v python3 >/dev/null 2>&1 && '
            'exec python3 -O -B -c "exec(bytes.fromhex(\'%s\').decode())" bscp-remote; '
            'for py in python2 python; do '
            'command -v "$py" >/dev/null 2>&1 && exec "$py" -O -B -c "%s" bscp-remote; '
            'done; ' % (hex_mt, script_py))
    wrapper = (
        py_branch +
        "command -v perl >/dev/null 2>&1 && "
        "exec perl -e 'eval pack(qq{H*}, q{%s})' bscp-remote; "
        'echo "bscp: no python or perl found on remote" >&2; exit 127'
    ) % hex_pl
    cmd += ['--', remote_host, wrapper]
    return cmd


def do_sync(local_file, remote_host, remote_file, mode,
            blocksize, section_size, hashname,
            start_offset=0, dry_run=False, block_count=0, allow_truncate=False, buffer=False,
            bwlimit=0, ssh_args=None):

    ssh_args = ssh_args or {}
    fn_bytes = remote_file.encode('utf-8')
    hn_bytes = hashname.encode('ascii')
    dig_size = hashlib.new(hashname).digest_size

    # Local is the destination on pull (we write blocks back into it) and the
    # source on push (read-only).  The remote uses the symmetric inverse —
    # see the open() in remote_script / remote_perl.
    open_mode = 'rb+' if mode == MODE_PULL else 'rb'
    try:
        f = open(local_file, open_mode)
    except (IOError, OSError) as e:
        # Py2 raises IOError, Py3 raises OSError (IOError is an alias).
        raise RuntimeError('Cannot open local file %s: %s' % (local_file, e))
    with f:
        f.seek(0, 2)
        local_size = f.tell()
        # wire_size is what we ship in the header — the source size as seen
        # by the protocol, after -B capping.  The unmodified local_size is
        # still used below for the authoritative truncate check.
        wire_size = local_size
        if block_count > 0:
            wire_size = min(wire_size, start_offset + block_count * blocksize)

        cmd  = build_ssh_cmd(remote_host, ssh_args)
        proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        io   = IOCounter(proc.stdout, proc.stdin, bwlimit=bwlimit)

        # The remote runs its own size-mismatch check on the wire-side `size`,
        # which is wire_size (the -B-capped local size) — so when -B is in use
        # the remote would falsely refuse.  The client's check below is
        # authoritative (it sees the real local_size and the requested limit),
        # so set the wire-side ALLOW_TRUNCATE bit whenever -B caps wire_size.
        wire_truncate = allow_truncate or block_count > 0
        mode_byte = mode | (ALLOW_TRUNCATE if wire_truncate else 0)

        def fail(msg):
            # Closing stdout too unblocks the remote if it is mid-stream on
            # stdout.write() (e.g. streaming hashes after we told it
            # ALLOW_TRUNCATE on the wire but then refuse on our side); without
            # it proc.wait() would deadlock against the remote's blocked write.
            _shutdown_proc(proc)
            raise RuntimeError(msg)

        def lost(msg):
            # Handshake-stage connection failure: the SSH pipe never produced a
            # valid reply (no route to host, dropped link, dead peer).  Unlike
            # fail(), raise ConnectionLost so the --retries loop re-establishes
            # the connection — a transient outage between attempts (the link was
            # up on a prior attempt) must not be mistaken for a permanent error.
            # Exception: rc 127 is our SSH wrapper's "no python or perl found on
            # remote", a genuinely fatal config error that retrying cannot fix.
            _shutdown_proc(proc)
            if proc.returncode == 127:
                raise RuntimeError(msg)
            raise ConnectionLost(start_offset)

        try:
            io.write(struct.pack(HEADER_FMT,
                                 wire_size, blocksize, section_size, start_offset,
                                 len(fn_bytes), len(hn_bytes), mode_byte))
            io.write(fn_bytes)
            io.write(hn_bytes)

            if io.read(dig_size) != hashlib.new(hashname, fn_bytes).digest():
                lost('Remote script failed to execute')
            io.write(b'go')

            size_bytes = io.read(8)
            if len(size_bytes) < 8:
                lost('Remote closed connection during handshake (got %d of 8 size bytes)'
                     % len(size_bytes))
        except (_BrokenPipeError, _ConnectionResetError):
            # A pipe error while still shipping the header is the same transient
            # connection loss, just surfaced on the write side; route it through
            # the same retry path.  (No IOTimeout here: the Py2 client has no
            # --io-timeout watchdog, so that exception never arises.)
            lost('Connection lost during handshake')
        (remote_size,) = struct.unpack('<Q', size_bytes)
        if remote_size == 0:
            fail('Remote file not found or inaccessible')

        # In push, remote is destination; in pull, remote is source.
        if mode == MODE_PUSH:
            src_label, src_size, dst_label, dst_size = 'local',  local_size,  'remote', remote_size
        else:
            src_label, src_size, dst_label, dst_size = 'remote', remote_size, 'local',  local_size

        # -B caps the source for the truncate check too: the user only wants
        # to sync up to that much, so a smaller destination is only "truncating"
        # if it can't fit even the requested amount.  An overshoot (-B asks for
        # more than the source has) is just a warning.
        eff_src = src_size
        if block_count > 0:
            requested = start_offset + block_count * blocksize
            if requested > src_size:
                report('\nWarning: -B requests up to %s but %s source has only %s; '
                       'syncing only what is available\n' % (
                       format_size(requested, floor=True),
                       src_label, format_size(src_size, floor=True)))
            else:
                eff_src = requested

        if dst_size < eff_src:
            if not allow_truncate:
                fail('%s destination size %d (%d blocks) < %s source size %d (%d blocks); '
                     'override with --allow-truncate' % (
                     dst_label, dst_size, dst_size // blocksize,
                     src_label, eff_src,  eff_src // blocksize))
            dst_h = format_size(dst_size, floor=True)
            report('\nWarning: %s destination (%s) smaller than %s source (%s); '
                   'syncing only first %s\n' % (
                   dst_label, dst_h, src_label,
                   format_size(eff_src, floor=True), dst_h))

        sync_size = min(wire_size, remote_size)
        if sync_size == 0:
            # Server has already exited (matching `if sync_size == 0: sys.exit(0)`).
            _shutdown_proc(proc)
            return io.in_total, io.out_total, 0, 0, 0
        eff_section    = section_size or sync_size
        total_blocks   = max(1, (sync_size + blocksize - 1) // blocksize)
        skipped_blocks = start_offset // blocksize
        total_scanned  = skipped_blocks

        total_diffs         = 0
        total_written       = 0
        total_blocks_copied = 0
        last_sec_start      = start_offset
        # One-element list cells so closures can mutate without `nonlocal`
        # (Python 2 has no nonlocal keyword).  Each cell holds one value
        # accessed as `var[0]`.
        t_last_progress     = [0.0]

        # Unified ETA state — tracked across sections so the estimate folds
        # both scan and copy work into one time budget.  Scan extrapolates
        # remaining blocks at the observed scan rate; copy extrapolates the
        # expected diff volume in the unscanned tail (using the diff-density
        # observed so far) at the observed copy rate.  No regime flip at the
        # phase A → phase B boundary; ETA is stable across sections.
        t_overall_start = time()
        total_scan_time = 0.0
        total_copy_time = 0.0

        # Pre-copy bootstrap: copy throughput is typically 1–2 orders of
        # magnitude below scan throughput.  Used until the first phase B
        # produces a real copy-rate measurement.
        BOOTSTRAP_COPY_RATIO = 64.0

        # Minimum cumulative scan/copy time before the rate is considered
        # reliable enough to drop the `~` bootstrap marker.  The very first
        # tick fires after a single block in microseconds, so the rate is
        # noise-dominated; before this threshold we display `~m:ss` to
        # signal low confidence.  Copy also has a byte-volume gate
        # (RATE_RELIABLE_COPY_BYTES) so fast transfers (localhost, 10 GbE)
        # — where the whole copy can finish in under a second — still seed
        # the EMA from real data instead of being marked bootstrap-quality
        # forever.
        RATE_RELIABLE_SECS       = 5.0
        RATE_RELIABLE_COPY_BYTES = 8 * 1024 * 1024

        # Diff-density EMA: per-section fraction of differing blocks,
        # smoothed across completed sections.  α=0.4 balanced.
        DIFF_EMA_ALPHA = 0.4
        ema_diff_frac  = None

        # Scan/copy rate EMAs — time-weighted to track real throughput
        # changes mid-run (HDD outer↔inner geometry, varying compressibility
        # under -C, network jitter).  RATE_EMA_TAU is the exponential decay
        # constant in seconds-of-phase-time; with REPORT_INTERVAL=0.5s, τ=5s
        # gives an effective ~10-sample window — long enough to ignore
        # per-tick noise, short enough to react to genuine rate changes
        # within a few seconds.  Rates are bootstrapped from the cumulative
        # average at the first sample where the phase has accumulated at
        # least RATE_RELIABLE_SECS of time, which sidesteps the bogus
        # first-block measurement entirely.
        # See t_last_progress above for the list-cell pattern.
        RATE_EMA_TAU         = 5.0
        ema_scan_rate        = [None]              # blocks/sec
        ema_copy_rate        = [None]              # bytes/sec
        rate_prev_scan_secs  = [0.0]
        rate_prev_scanned    = [skipped_blocks]
        rate_prev_copy_secs  = [0.0]
        rate_prev_written    = [0]

        # ETA display damping: the underlying estimate is recomputed every
        # progress tick into a heavily-weighted EMA (`eta_smoothed`), but
        # the *displayed* value is only re-snapped from `eta_smoothed` every
        # ETA_REFRESH_INTERVAL seconds and held constant between snaps.
        # The user sees the ETA stay still for a full refresh window and
        # then update in one step — explicitly chosen over per-tick (or
        # per-second) countdown so the number on screen does not appear to
        # drift independently of measured progress.  Damping is applied
        # from the start (not gated on `confident`), so an unseeded early
        # estimate never gets a chance to flicker — for the first
        # ETA_WARMUP_SECS of the run, display_eta() returns None and the
        # caller omits the `(ETA)` field from the progress line entirely,
        # which gives the rate EMAs time to seed from real data before any
        # number is shown.
        ETA_REFRESH_INTERVAL = 10.0
        ETA_WARMUP_SECS      = 20.0
        # Confidence shortcut: once at least CONFIDENT_SCAN_FRAC of the
        # blocks-still-to-scan have been scanned, treat the scan-rate EMA
        # as authoritative even if the copy-rate EMA never seeds (e.g.
        # zero diffs found, so no copy happens).  Without this gate, a
        # clean run with no differing blocks would keep the `~` bootstrap
        # marker for its entire duration.
        CONFIDENT_SCAN_FRAC  = 0.05
        # See t_last_progress above for the list-cell pattern.
        eta_smoothed         = [None]
        eta_displayed        = [None]
        eta_displayed_at     = [0.0]

        def display_eta(fresh_eta, now):
            if now - t_overall_start < ETA_WARMUP_SECS:
                return None
            if eta_smoothed[0] is None:
                eta_smoothed[0] = fresh_eta
            eta_smoothed[0] = (99.0 * (eta_smoothed[0] - REPORT_INTERVAL) + fresh_eta) / 100.0
            if now - eta_displayed_at[0] >= ETA_REFRESH_INTERVAL:
                eta_displayed[0] = eta_smoothed[0]
                eta_displayed_at[0] = now
            return eta_displayed[0]

        def update_scan_rate_ema(scan_secs_now):
            if scan_secs_now < RATE_RELIABLE_SECS:
                return
            if ema_scan_rate[0] is None:
                # Seed from cumulative — bypasses first-tick noise entirely.
                ema_scan_rate[0]       = (total_scanned - skipped_blocks) / scan_secs_now
                rate_prev_scan_secs[0] = scan_secs_now
                rate_prev_scanned[0]   = total_scanned
                return
            dt    = scan_secs_now - rate_prev_scan_secs[0]
            delta = total_scanned - rate_prev_scanned[0]
            if dt <= 0 or delta <= 0:
                return
            inst          = delta / dt
            w             = 1.0 - math.exp(-dt / RATE_EMA_TAU)
            ema_scan_rate[0] = (1 - w) * ema_scan_rate[0] + w * inst
            rate_prev_scan_secs[0] = scan_secs_now
            rate_prev_scanned[0]   = total_scanned

        def update_copy_rate_ema(copy_secs_now, written_now):
            if written_now <= 0:
                return
            if (copy_secs_now < RATE_RELIABLE_SECS
                    and written_now < RATE_RELIABLE_COPY_BYTES):
                return
            if ema_copy_rate[0] is None:
                ema_copy_rate[0]       = written_now / copy_secs_now
                rate_prev_copy_secs[0] = copy_secs_now
                rate_prev_written[0]   = written_now
                return
            dt    = copy_secs_now - rate_prev_copy_secs[0]
            delta = written_now - rate_prev_written[0]
            if dt <= 0 or delta <= 0:
                return
            inst          = delta / dt
            w             = 1.0 - math.exp(-dt / RATE_EMA_TAU)
            ema_copy_rate[0] = (1 - w) * ema_copy_rate[0] + w * inst
            rate_prev_copy_secs[0] = copy_secs_now
            rate_prev_written[0]   = written_now

        def estimate_eta(now, live_diffs=0, live_written=0,
                         scan_active_since=None, copy_active_since=None,
                         sec_blocks_scanned=0, sec_blocks_total=0):
            scan_secs = total_scan_time
            if scan_active_since is not None:
                scan_secs += now - scan_active_since

            scan_blocks_done = max(1, total_scanned - skipped_blocks)
            if ema_scan_rate[0] is not None:
                scan_rate = ema_scan_rate[0]
            elif scan_secs > 0:
                scan_rate = scan_blocks_done / scan_secs
            else:
                scan_rate = 0.0
            unscanned = max(0, total_blocks - total_scanned)
            remaining_scan_time = unscanned / scan_rate if scan_rate > 0 else 0.0

            # Dry-run does not transfer any blocks — the copy component is
            # zero by definition, so the ETA collapses to scan-only.
            if dry_run:
                return remaining_scan_time

            copy_secs = total_copy_time
            if copy_active_since is not None:
                copy_secs += now - copy_active_since

            written_cum = total_written + live_written
            if ema_copy_rate[0] is not None:
                copy_rate = ema_copy_rate[0]
            elif written_cum > 0 and copy_secs > 0:
                copy_rate = written_cum / copy_secs
            else:
                copy_rate = max(1.0, scan_rate * blocksize / BOOTSTRAP_COPY_RATIO)

            # Diff-density estimate.  When called from phase A, the caller
            # passes section progress; the in-section running fraction is
            # blended with the completed-section EMA, weighted by section
            # completion.  When called from phase B (sec_blocks_total == 0)
            # the section's scan is already folded into the EMA, so we use
            # it directly.
            if sec_blocks_scanned > 0:
                sec_so_far = live_diffs / sec_blocks_scanned
            else:
                sec_so_far = 0.0
            if ema_diff_frac is None:
                diff_frac = sec_so_far
            elif sec_blocks_total <= 0:
                diff_frac = ema_diff_frac
            else:
                w_running = sec_blocks_scanned / sec_blocks_total
                diff_frac = w_running * sec_so_far + (1.0 - w_running) * ema_diff_frac

            diffs_cum        = total_diffs + live_diffs
            est_future_diffs = diff_frac * unscanned
            est_copy_bytes   = (diffs_cum + est_future_diffs) * blocksize
            remaining_copy   = max(0.0, est_copy_bytes - written_cum)

            remaining_copy_time = remaining_copy / copy_rate if copy_rate > 0 else 0.0
            return remaining_scan_time + remaining_copy_time

        try:
            pos = start_offset
            while pos < sync_size:
                sec_end        = min(pos + eff_section, sync_size)
                last_sec_start = pos

                # ── Phase A: hash exchange ────────────────────────────
                diff_blocks    = []   # push: [(wpos, block)]
                diff_positions = []   # pull: [rpos]

                t_phase_a_start = time()
                f.seek(pos)
                p = pos
                while p < sec_end:
                    bl         = min(blocksize, sec_end - p)
                    block      = f.read(bl)
                    local_dig  = hashlib.new(hashname, block).digest()
                    remote_dig = io.read(dig_size)
                    if len(remote_dig) < dig_size:
                        raise EOFError('SSH pipe closed during phase-A hash read '
                                       '(got %d of %d bytes)' % (len(remote_dig), dig_size))
                    if local_dig != remote_dig:
                        if mode == MODE_PUSH and buffer:
                            diff_blocks.append((p, block))
                        else:
                            diff_positions.append(p)
                    p             += bl
                    total_scanned += 1
                    now = time()
                    if now - t_last_progress[0] >= REPORT_INTERVAL:
                        t_last_progress[0] = now
                        sec_diffs    = len(diff_blocks) + len(diff_positions)
                        ndiff        = total_diffs + sec_diffs
                        pct_done     = total_scanned * 100.0 / total_blocks
                        newly        = total_scanned - skipped_blocks
                        pct_diff     = ndiff * 100.0 / max(1, newly)
                        elapsed      = now - t_overall_start
                        scan_secs    = total_scan_time + (now - t_phase_a_start)
                        speed        = newly * blocksize / (scan_secs * 1024**2 + 1)
                        update_scan_rate_ema(scan_secs)
                        # Copy phase is not currently running, but earlier
                        # sections' phase B may have written bytes — sample
                        # the EMA from the cumulative totals so it can seed
                        # without waiting for the next copy tick.
                        update_copy_rate_ema(total_copy_time, total_written)
                        # `~` marker means "low confidence" — rate samples
                        # are still bootstrap-quality.  Drops once the rate
                        # EMAs are seeded (or scan-only is seeded under
                        # --dry-run), or — for runs with zero diffs found —
                        # once CONFIDENT_SCAN_FRAC of the file has been
                        # scanned and the absence of diffs can be trusted.
                        scan_frac    = (total_scanned - skipped_blocks) / max(
                                           1, total_blocks - skipped_blocks)
                        confident    = ema_scan_rate[0] is not None and (
                                           dry_run or ema_copy_rate[0] is not None
                                           or scan_frac >= CONFIDENT_SCAN_FRAC)
                        sec_blk_done = (p   - pos + blocksize - 1) // blocksize
                        sec_blk_tot  = (sec_end - pos + blocksize - 1) // blocksize
                        eta          = display_eta(estimate_eta(now,
                                                    live_diffs=sec_diffs,
                                                    scan_active_since=t_phase_a_start,
                                                    sec_blocks_scanned=sec_blk_done,
                                                    sec_blocks_total=sec_blk_tot),
                                                    now)
                        eta_str  = '' if confident or eta is None else '~'
                        if eta is not None:
                            eta_str = " (%s%s)" %(eta_str, format_time(eta, round_eta=True))
                        scanned_bytes = min(total_scanned * blocksize, sync_size)
                        report('\rscan %s/%s (%.1f%%) diff blocks=%d (%.1f%%) (%.0f MiB/s) %s%s ' % (
                            format_size(scanned_bytes, floor=True),
                            format_size(sync_size, floor=True),
                            pct_done, ndiff, pct_diff, speed,
                            format_time(elapsed), eta_str))

                total_scan_time += time() - t_phase_a_start
                sec_blocks_total = (sec_end - pos + blocksize - 1) // blocksize
                sec_diff_count   = len(diff_blocks) + len(diff_positions)
                sec_frac         = sec_diff_count / max(1, sec_blocks_total)
                ema_diff_frac    = (sec_frac if ema_diff_frac is None
                                    else DIFF_EMA_ALPHA * sec_frac
                                       + (1 - DIFF_EMA_ALPHA) * ema_diff_frac)
                total_diffs     += sec_diff_count

                # ── Phase B: transfer ─────────────────────────────────
                t_phase_b_start    = time()
                written            = 0
                section_diff_count = len(diff_blocks) + len(diff_positions)

                def show_copy_progress(wpos, blocks_done):
                    now = time()
                    if now - t_last_progress[0] < REPORT_INTERVAL:
                        return
                    t_last_progress[0] = now
                    sec_elapsed = max(1e-6, now - t_phase_b_start)
                    speed       = written / (sec_elapsed * 1024 + 1)
                    copy_secs   = total_copy_time + (now - t_phase_b_start)
                    update_copy_rate_ema(copy_secs, total_written + written)
                    scan_frac   = (total_scanned - skipped_blocks) / max(
                                      1, total_blocks - skipped_blocks)
                    confident   = ema_scan_rate[0] is not None and (
                                      ema_copy_rate[0] is not None
                                      or scan_frac >= CONFIDENT_SCAN_FRAC)
                    eta         = display_eta(estimate_eta(now,
                                               live_written=written,
                                               copy_active_since=t_phase_b_start),
                                               now)
                    eta_str  = '' if confident or eta is None else '~'
                    if eta is not None:
                        eta_str = " (%s%s)" %(eta_str, format_time(eta, round_eta=True))
                    pct_pos     = wpos * 100.0 / sync_size
                    pct_sec     = blocks_done * 100.0 / max(1, section_diff_count)
                    cum_blocks  = total_blocks_copied + blocks_done
                    report('\rcopy %s/%s (%.1f%%) block %d/%d (%.1f%%) (%.0f KiB/s) %s%s ' % (
                        format_size(wpos, floor=True),
                        format_size(sync_size, floor=True),
                        pct_pos, cum_blocks, total_diffs, pct_sec,
                        speed, format_time(now - t_overall_start),
                        eta_str))

                if mode == MODE_PUSH and not buffer:
                    io.write(struct.pack('<Q', 0 if dry_run else len(diff_positions)))
                    if not dry_run:
                        for i, wpos in enumerate(diff_positions):
                            bl = min(blocksize, sync_size - wpos)
                            f.seek(wpos)
                            io.write(struct.pack('<Q', wpos))
                            io.write(f.read(bl))
                            written += bl
                            show_copy_progress(wpos, i + 1)

                elif mode == MODE_PUSH:
                    io.write(struct.pack('<Q', 0 if dry_run else len(diff_blocks)))
                    if not dry_run:
                        for i, (wpos, wblock) in enumerate(diff_blocks):
                            io.write(struct.pack('<Q', wpos))
                            io.write(wblock)
                            written += len(wblock)
                            show_copy_progress(wpos, i + 1)

                else:  # pull — windowed to avoid two-pipe deadlock
                    io.write(struct.pack('<Q', 0 if dry_run else len(diff_positions)))
                    if not dry_run:
                        blocks_done = 0
                        for batch_start in range(0, len(diff_positions), PULL_WINDOW):
                            window = diff_positions[batch_start:batch_start + PULL_WINDOW]
                            for rpos in window:
                                io.write(struct.pack('<Q', rpos))
                            for rpos in window:
                                bl    = min(blocksize, sync_size - rpos)
                                block = io.read(bl)
                                if len(block) < bl:
                                    raise EOFError('SSH pipe closed during pull phase-B block read '
                                                   '(got %d of %d bytes at offset %d)'
                                                   % (len(block), bl, rpos))
                                f.seek(rpos)
                                f.write(block)
                                written      += bl
                                blocks_done  += 1
                                show_copy_progress(rpos, blocks_done)

                total_copy_time     += time() - t_phase_b_start
                total_written       += written
                total_blocks_copied += section_diff_count
                pos = sec_end

        except _PIPE_ERRORS:
            # Closing stdout in addition to stdin unblocks the remote if it
            # was stuck mid-write — same rationale as fail() above.
            _shutdown_proc(proc)
            # `raise X from Y` is Py3-only; cause chain is dropped on Py2.
            raise ConnectionLost(last_sec_start)
        except KeyboardInterrupt:
            # Mirror the BrokenPipeError path: close stdout too, otherwise a
            # remote blocked mid-write (pull phase B, or any phase A) keeps
            # proc.wait() hanging forever and the user sees no resume hint.
            _shutdown_proc(proc)
            raise ConnectionLost(last_sec_start, interrupted=True)

        _shutdown_proc(proc)

    return io.in_total, io.out_total, sync_size, total_diffs, total_written


if __name__ == '__main__':
    parser = ArgumentParser(
        description='Binary Secure Copy — push or pull a local/remote file or block device, '
                    'transferring only changed blocks over SSH.',
        usage='%(prog)s [options] SRC DST\n'
              '  push:  bscp [options] local_file   HOST:remote_file\n'
              '  pull:  bscp [options] HOST:remote_file   local_file')
    transfer = parser.add_argument_group('transfer tuning')
    transfer.add_argument('-b', '--block-size',   default=DEFAULT_BLOCK,
                        type=parse_size, metavar='SIZE',
                        help='block size (default 64K; supports K/M/G/T suffix. Increasing block '
                             'size, e.g. to 256K, might be beneficial to throughput on fast devices.)')
    transfer.add_argument('-s', '--section-size', default=DEFAULT_SECTION,
                        type=parse_size, metavar='SIZE',
                        help='section size (default 10G; bounds memory use per pass)')
    transfer.add_argument('-a', '--algorithm',    default='sha256',
                        type=parse_algorithm, metavar='ALGO',
                        help=algorithm_help())
    transfer.add_argument('-B', '--block-count',      default=0, type=parse_block_count, metavar='N',
                        help='limit sync to first N blocks (default 0 = no limit). '
                             'A K/M/G/T suffix interprets the value as bytes, '
                             'rounded up to a whole number of blocks.')

    perf = parser.add_argument_group('performance')
    perf.add_argument('--buffer',                     action='store_true',
                        help='push: buffer differing blocks in memory during '
                             'transfer instead of re-reading them from disk. '
                             'Higher memory use, but only one disk read pass needed. Experimental.')

    ssh = parser.add_argument_group('SSH connection')
    ssh.add_argument('-p', '--port',            default=22, type=int, metavar='PORT',
                        help='SSH port (default 22)')
    ssh.add_argument('-i', '--identity',        metavar='FILE',
                        help='SSH identity file (-i FILE)')
    ssh.add_argument('-o', '--ssh-opt',         action='append', metavar='OPT',
                        help='extra SSH option, repeatable (passed as -o OPT)')
    ssh.add_argument('-C', '--compress',        action='store_true',
                        help='enable SSH compression')

    resilience = parser.add_argument_group('resilience and resume')
    resilience.add_argument('-r', '--resume-from',  default=0,
                        type=parse_resume_from, metavar='OFFSET',
                        help='resume from byte offset (K/M/G/T suffix accepted), '
                             'or NN[.N]%% of the local file size; rounded down '
                             'to section boundary')
    resilience.add_argument('-R', '--retries',  default=DEFAULT_RETRIES, type=int, metavar='N',
                        help='auto-retry attempts on connection failure (default %d; 0 disables)' % DEFAULT_RETRIES)
    resilience.add_argument('--bwlimit',        default=0, type=parse_size, metavar='RATE',
                        help='limit network I/O to RATE bytes per second (K/M/G/T suffix accepted); '
                             'governs both directions combined. Default 0 = unlimited.')

    safety = parser.add_argument_group('safety')
    safety.add_argument('-N', '--dry-run',      action='store_true',
                        help='count differences only; do not update destination')
    safety.add_argument('--allow-truncate',           action='store_true',
                        help='allow destination to be smaller than source; '
                             'syncs only the bytes that fit. '
                             'WARNING: data beyond destination size is not copied.')

    output = parser.add_argument_group('output')
    output.add_argument('-q', '--quiet',              action='store_true',
                        help='suppress scan/copy progress lines; errors and warnings are still shown')
    output.add_argument('--batch',                    action='store_true',
                        help='suppress all stderr output; use exit status to detect errors (implies -q). '
                             'Exit 0=success, 1=fatal error, 2=bad arguments, '
                             '3=connection lost, 130=interrupted. '
                             'Note: the exit code cannot convey the resume offset; '
                             'callers that need to resume should use -q instead and '
                             'parse the "Resume with:" line from stderr.')

    parser.add_argument('src', help='source — local path or HOST:path')
    parser.add_argument('dst', help='destination — HOST:path or local path')

    # BSCP_OPTIONS supplies default command-line options for per-host tuning
    # (e.g. `-b 192k -s 64M`), parsed as a shell word list and placed BEFORE
    # the real argv so an explicit command-line option overrides the env
    # default (argparse keeps the last occurrence).  Options only — not
    # positionals.  Distinct from the BSCP_FORCE_* test hooks, which select
    # the remote path.
    try:
        env_opts = shlex.split(os.environ.get('BSCP_OPTIONS', ''))
    except ValueError as exc:
        parser.error('BSCP_OPTIONS: %s' % exc)

    if len(sys.argv) == 1 and not env_opts:
        parser.print_usage(sys.stderr)
        sys.stderr.write('Use -h or --help for full option list.\n')
        sys.exit(2)

    args = parser.parse_args(env_opts + sys.argv[1:])
    quiet = args.quiet or args.batch
    batch = args.batch

    src, dst = args.src, args.dst
    if ':' in src and ':' not in dst:
        mode = MODE_PULL
        remote_host, remote_file = src.split(':', 1)
        local_file = dst
    elif ':' in dst and ':' not in src:
        mode = MODE_PUSH
        local_file = src
        remote_host, remote_file = dst.split(':', 1)
    else:
        parser.error('exactly one of SRC or DST must be HOST:path')

    blocksize    = args.block_size
    section_size = args.section_size
    hashname     = args.algorithm
    dry_run      = args.dry_run
    # parse_resume_from() returns a negative permille sentinel for the NN%
    # form; resolve against the local file size before section-boundary
    # rounding.  device_size() (seek-to-end, *not* os.path.getsize, which
    # reports 0 for block devices and would silently collapse any NN% to
    # offset 0).  If the file is missing, fall through with offset 0 — the
    # friendlier 'Cannot open local file ...' error fires inside do_sync.
    if args.resume_from < 0:
        try:
            local_size_for_pct = device_size(local_file)
        except (IOError, OSError):
            local_size_for_pct = 0
        resume_offset = (-args.resume_from) * local_size_for_pct // 1000
    else:
        resume_offset = args.resume_from
    # section_size == 0 means whole-file-in-one-section, in which case the
    # only valid resume boundary is start_offset itself.
    start_offset = (resume_offset // section_size) * section_size if section_size else resume_offset
    ssh_args     = vars(args)

    # parse_block_count() returns a negative sentinel for byte-suffixed -B
    # values; resolve to a block count now that blocksize is known.
    if args.block_count < 0:
        args.block_count = (-args.block_count + blocksize - 1) // blocksize

    # For push with --buffer, cap section size at half of available memory
    # so that the diff_blocks list (one full block per diff) cannot exhaust RAM.
    # Without --buffer (or while pulling), only 8-byte positions are stored, so no cap needed.
    buffer = args.buffer
    if mode == MODE_PUSH and buffer:
        avail = available_memory()
        if avail < MIN_MEMORY_PUSH:
            buffer = False
            report('\nNote: dropping --buffer mode due to low available memory\n')
        else:
            mem_limit = max(avail // 2, blocksize)
            # section_size == 0 means "whole file as one section" — also clamp.
            if section_size == 0 or section_size > mem_limit:
                section_size = mem_limit
                report('\nNote: section size reduced to %s (half of available memory)\n'
                        % format_size(section_size, floor=True))

    resume_cmd = lambda offset: build_resume_cmd(
        sys.argv[0], args, src, dst, offset)
    # The offset this run actually started from (after NN% resolution and
    # section-boundary rounding).  An abort whose resume offset still equals
    # this means no section completed, so a "-r OFFSET" resume command would
    # just reproduce the original invocation — we print the bare status line
    # instead.  Conversely, a printed resume command signals that some work is
    # banked and need not be repeated.  Compared against the *initial* offset,
    # not the retry-mutated start_offset, so progress from an earlier attempt
    # is still surfaced after later attempts stall.
    initial_offset = start_offset
    progressed     = lambda offset: offset != initial_offset

    t_start  = time()
    attempt  = 0

    try:
        while True:
            try:
                (in_total, out_total, sync_size,
                 total_diffs, total_written) = do_sync(
                    local_file, remote_host, remote_file,
                    mode, blocksize, section_size, hashname,
                    start_offset=start_offset, dry_run=dry_run,
                    block_count=args.block_count, allow_truncate=args.allow_truncate,
                    buffer=buffer, bwlimit=args.bwlimit, ssh_args=ssh_args)
                break
            except ConnectionLost as exc:
                if attempt < args.retries and not exc.interrupted:
                    delay = min(2 ** attempt, 60)
                    if progressed(exc.offset):
                        report('\rConnection lost — retrying (%d/%d) in %ds...\n'
                                'Resume manually with:\n  %s\n' % (
                                attempt + 1, args.retries, delay,
                                resume_cmd(exc.offset)))
                    else:
                        report('\rConnection lost — retrying (%d/%d) in %ds...\n' % (
                                attempt + 1, args.retries, delay))
                    start_offset = exc.offset
                    attempt += 1
                    sleep(delay)
                else:
                    label = 'Interrupted' if exc.interrupted else 'Connection lost'
                    if progressed(exc.offset):
                        report('\n%s - Resume with:\n  %s\n' % (
                            label, resume_cmd(exc.offset)))
                    else:
                        report('\n%s\n' % label)
                    sys.exit(130 if exc.interrupted else 3)
            except RuntimeError as exc:
                report('\nError: %s\n' % exc)
                sys.exit(1)
    except KeyboardInterrupt:
        if progressed(start_offset):
            report('\nInterrupted - Resume with:\n  %s\n' % resume_cmd(start_offset))
        else:
            report('\nInterrupted\n')
        sys.exit(130)

    elapsed  = time() - t_start
    xfer     = in_total + out_total
    speedup  = sync_size / (xfer + 1)

    quiet = False # show final report unless --batch is active
    if dry_run:
        dsm = "+/- %s of " %format_size(total_diffs * blocksize, floor=True) if total_diffs else ""
        report('\rdry-run: diff blocks=%d (%s%s) (%.1f%%) (%s)\n' % (
            total_diffs, dsm, format_size(sync_size, floor=True),
            (100.0 * total_diffs * blocksize) / sync_size, format_time(elapsed)))
    else:
        report('\rin=%s out=%s size=%s diff blocks=%d speedup=%.2f (%s)\n' % (
            format_size(in_total,  floor=True),
            format_size(out_total, floor=True),
            format_size(sync_size, floor=True),
            total_diffs, speedup, format_time(elapsed)))

    if args.block_count:
        report('Continue with:\n  %s\n' % build_resume_cmd(
            sys.argv[0], args, src, dst, sync_size, block_count=0))

# The remote source literals (remote_script, remote_script_mt, remote_perl)
# are tab-indented; ts=4 renders those embedded tabs at the same visual width
# as the surrounding 4-space client code.  See the note above remote_script.
# vim: set ts=4: