FileSync¶

class rmote.tools.file_sync.FileSync[source]¶

Bases: Tool

Synchronize regular files through an open asynchronous Protocol.

Call await FileSync.upload(protocol, local_path, remote_path) or await FileSync.download(protocol, remote_path, local_path) directly. These methods coordinate both machines locally; do not pass them as the tool argument to protocol(...). The target needs Python but no rmote installation. Paths are interpreted on their respective sides; parent directories must already exist and symlinks/special files are rejected.

The sender hashes a block with SHA-256; the receiver compares its own block at that offset and requests content only on mismatch. Blocks default to 1 MiB and are negotiated one at a time. Memory is bounded by block size, but network latency limits throughput. Insertions can shift later block boundaries and cause large retransfers.

The receiver assembles a temporary file beside the destination, verifies the cumulative digest, flushes and fsyncs, then atomically replaces it. Existing destination mode/uid/gid are preserved, new files use 0600. Source timestamps, ACLs and extended attributes are not copied. Other hard links still point to the old file. Identical files retain inode and timestamps, though comparison reads both files and writes a temporary copy. Allow free space for the complete result, even if only one block differs.

Keep both files stable: version checks detect ordinary concurrent changes but do not provide locking. Errors before replacement preserve the target. Cancellation waits for the current RPC before cleanup and can therefore wait on a stalled connection. An in-flight final replacement cannot be undone. Lost connections or abrupt process termination may leave a .<filename>.rmote-* temporary file; retry after an ambiguous final RPC failure to establish the resulting state. There is no resume or directory synchronization, and no guarantee of durability across power loss.

Example using a real local subprocess (no SSH server required):

>>> import asyncio
>>> import sys
>>> from pathlib import Path
>>> from tempfile import TemporaryDirectory
>>> from rmote.protocol import Protocol
>>> async def example():
...     with TemporaryDirectory() as directory:
...         root = Path(directory)
...         source, target, copy = (root / n for n in ("source", "target", "copy"))
...         source.write_bytes(b"hello")
...         process = await asyncio.create_subprocess_exec(
...             sys.executable, "-qui", stdin=asyncio.subprocess.PIPE,
...             stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL,
...         )
...         try:
...             protocol = await Protocol.from_subprocess(process)
...             async with protocol:
...                 first = await FileSync.upload(protocol, source, target)
...                 again = await FileSync.upload(protocol, source, target)
...                 await FileSync.download(protocol, target, copy)
...                 return first.changed, again.changed, copy.read_bytes()
...         finally:
...             if process.returncode is None:
...                 process.terminate()
...             await process.wait()
>>> asyncio.run(example())
(True, False, b'hello')

With SSH, enter async with await Protocol.from_ssh("user@host") as protocol and use the same upload/download calls. The SSH protocol owns its process; the local subprocess example explicitly owns and reaps its child.

async static download(protocol, remote_path, local_path, *, block_size=1048576)[source]¶

Synchronize remote source content into a local destination.

Parameters:

protocol (Callable[..., Awaitable[Any]]) – An open async Protocol, already entered with async with.
remote_path (Text | Path) – Existing remote source file; relative to target cwd.
local_path (Text | Path) – Local destination file; relative to local cwd. Its parent must exist. A missing file is created with mode 0600.
block_size (getint) – Bytes per block, 1 through 16 MiB; defaults to 1 MiB.

Return type:

SyncResult

Returns:

SyncResult with changed status and transferred/reused byte counts.

Raises:

ValueError – Invalid block size, special file, or failed integrity check.
RuntimeError – A file changed while being synchronized.
OSError – File access, temporary-file creation or installation fails.
ConnectionError – Closed connection; other transport errors propagate.
asyncio.CancelledError – Cancelled after the in-flight RPC and cleanup.

Uses the same exchange and atomic replacement as upload, with the remote side as sender. Call directly: await FileSync.download(protocol, src, dst).

async static transfer(protocol, local_path, remote_path, uploading, block_size)[source]¶

Return type:: SyncResult

async static upload(protocol, local_path, remote_path, *, block_size=1048576)[source]¶

Synchronize local source content into a remote destination.

Parameters:

protocol (Callable[..., Awaitable[Any]]) – An open async Protocol, already entered with async with.
local_path (Text | Path) – Existing local source file; relative to local cwd.
remote_path (Text | Path) – Remote destination file; relative to the target cwd. Its parent must exist. A missing file is created with mode 0600.
block_size (getint) – Bytes per block, 1 through 16 MiB; defaults to 1 MiB.

Return type:

SyncResult

Returns:

SyncResult with changed status and transferred/reused byte counts. Repeating the call with identical content returns changed=False.

Raises:

ValueError – Invalid block size, special file, or failed integrity check.
RuntimeError – A file changed while being synchronized.
OSError – File access, temporary-file creation or installation fails.
ConnectionError – The connection is closed. Transport failures may also propagate their original exception.
asyncio.CancelledError – Cancelled after the in-flight RPC and cleanup.

Uses the atomic replacement and metadata policy documented on FileSync.

Types¶

class rmote.tools.file_sync.SyncResult(changed, size, transferred, reused)[source]¶

Bases: object

Summary of a completed file synchronization.

changed¶: Whether the destination was created or replaced. False leaves its inode and timestamps unchanged; source metadata is not copied.

size¶: Final file size in bytes.

transferred¶: File-content bytes sent across the connection, before compression. Excludes signatures, RPC framing and transferred code.

reused¶: Bytes copied from the existing destination into the temporary file. transferred + reused == size even when only truncating.

Low-level API¶

class rmote.tools.file_sync.Session(path, receiving, block_size, size=0)[source]¶

Bases: object

One sender or receiver in the block exchange used by FileSync.

Opens a regular file and keeps at most one block of content in memory. A receiver also opens a temporary file beside the destination. Always call close() in a finally block, including after successful completion. Calls on one session must be sequential.

Parameters:

path (Text) – Source or destination path in this process’s filesystem.
receiving (bool) – True to assemble a destination, False to read a source.
block_size (getint) – Block size in bytes, from 1 through 16 MiB inclusive.
size (getint) – Expected final size for a receiver; ignored by a sender, whose size is read from the open source file.

Raises:

ValueError – Invalid block size, negative size, or a special file.
OSError – Opening the file fails, including missing source or parent, insufficient permissions, or a symlink at the supplied path.

step() advances the exchange and returns the message for the other session. A receiver returns SyncResult after verifying the complete file and committing it. Application code normally uses FileSync.

append(data)[source]¶

Append already verified bytes to a receiver’s temporary file.

Updates its position and cumulative SHA-256. Use step for normal exchange: this helper does not validate a block signature or count it as transferred/reused.

Return type:: None

check()[source]¶

Raise RuntimeError if the file’s identity, size or timestamps changed.

These checks detect ordinary concurrent writes; they do not lock the file or make a snapshot. Keep both files stable throughout the transfer.

Return type:: None

close()[source]¶

Close file descriptors and remove any remaining temporary file.

Does not install incomplete output or undo a completed replacement. Repeated calls are safe. Filesystem errors during cleanup propagate.

Return type:: None

finish(digest)[source]¶

Verify the receiver’s final SHA-256 and atomically install its output.

All bytes must have arrived and no requested block may be outstanding. An identical destination is left untouched. Otherwise flush and fsync the temporary file, preserve existing mode/uid/gid, and use os.replace. New files have mode 0600. The containing directory is not fsynced, so this is atomic replacement, not a power-loss durability guarantee.

Raises:

ValueError – Incomplete transfer, wrong digest or a sender session.
RuntimeError – The destination changed during transfer.
OSError – Flushing, metadata preservation or replacement fails.

Call close afterwards to release descriptors and any temporary file.

Return type:: SyncResult

step(message=None)[source]¶

Exchange a signature, match reply, or requested block.

Start the sender with None. It returns (length, sha256_digest). Feed this to the receiver: True means its old block matched and was copied; False requests the pending source bytes. Pass that reply to the sender. On False it returns those bytes; after writing and verifying them the receiver replies None. True or None advances the sender.

A zero-length signature ends the stream and carries the whole-file hash. The receiver then returns SyncResult. Stop exchanging messages at that point and close both sessions.

Raises:

ValueError – Invalid signature, unexpected/corrupt content, incomplete transfer, or mismatched whole-file digest.
RuntimeError – Either this file or its open descriptor changed.
OSError – A read, write, fsync, metadata update or replacement fails.

Return type:

Any