DigitalInfinity/clonetool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync parallen mit copy und windows-support

Browse Source
This commit is contained in:
Alexander Gabriel 2026-09-06 01:44:45 +02:00
parent 95783225d5
commit 20c9017d3b
14 changed files with 549 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1 +1,2 @@
/clonetool
/dist/

73
README.md
View File

@ -16,12 +16,20 @@ system `ssh` client for remote endpoints.
A block is only re-read once: it's hashed from the buffer it was read
into, and that same buffer is what gets sent on if it differs — never
read twice.
- **Source and destination are scanned in parallel.** The destination side
streams the hash of each block *as it computes it*, in block order; the
source side consumes that stream and immediately reads, compares and (if
it differs) sends that block. So the destination scan, the source scan
and the transfer of changed blocks all overlap — there is no "scan the
whole destination, then start" phase. The status line shows the transfer
progress with a trailing `scan H/M` until the destination fingerprint is
complete.
- **No state is kept between runs.** Every sync re-reads and re-hashes the
destination's *current* content and compares the source against that, so
a re-run only moves the blocks that actually differ and nothing needs to
be trusted from a previous run. (The destination side does the
destination hashing; in push mode it streams that table to the source,
in pull mode the destination agent hashes locally.)
be trusted from a previous run. (The destination side always does the
destination hashing — the remote `sink` in push mode, the local
destination agent in pull mode.)
- Bulk data goes **directly between source and destination**, not through
the manager. Each run tries:
1. **push** — the source agent connects straight to the destination
@ -66,10 +74,54 @@ system `ssh` client for remote endpoints.
CGO_ENABLED=0 go build -o clonetool .
```
Cross-compile for another OS/arch by setting `GOOS`/`GOARCH` (no cgo, so
these all work from any host):
```
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o clonetool .
CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -o clonetool.exe .
CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build -o clonetool-darwin .
```
`./build.sh` writes all three of the above (plus linux/amd64) into `dist/`.
Copy the resulting binary to the manager host. Source and destination
hosts get it automatically (see self-deploy above), or place it yourself
and point `--remote-bin` at it. The binary is architecture-specific —
cross-compile (`GOOS`/`GOARCH`) if your hosts differ.
cross-compile if your hosts differ. Self-deploy streams a POSIX shell
script over SSH, so a **Windows** host can be the manager or a local
endpoint but cannot be an automatic deploy target — put `clonetool.exe`
on it yourself and point `--remote-bin` at it.
## Windows
clonetool runs on Windows and can clone **physical drives and volumes**,
not just files:
```
# Whole disk to an image file
clonetool sync --source \\.\PhysicalDrive2 --dest D:\backup\disk2.img
# Image file back onto a disk (must not be larger than the disk)
clonetool sync --source D:\backup\disk2.img --dest \\.\PhysicalDrive2
# A single volume
clonetool sync --source \\.\E: --dest \\.\F:
```
- Raw-disk paths are `\\.\PhysicalDrive<n>` (whole disk) or `\\.\<X>:` (a
volume). `\\?\` also works. Forward slashes are accepted.
- **Run from an elevated (Administrator) console** to open a raw disk.
There is no `sudo` fallback on Windows; `--sudo` is ignored. A
permission error tells you to elevate.
- A raw-disk **destination** should have no mounted filesystem in use
(take the disk offline in Disk Management, or target a volume that
nothing else has open) — Windows blocks writes to a disk region owned by
a mounted volume. Reading a live disk as the **source** is fine.
- Raw-disk I/O must be sector-aligned. `--block-size` must be a multiple
of the drive's sector size (512 or 4096); the default 4M is. clonetool
handles the final partial block itself.
- Device size is read with `IOCTL_DISK_GET_LENGTH_INFO`.
## Usage
@ -111,16 +163,19 @@ the self-deploy check).
## Caveats
- Block-device size detection (`BLKGETSIZE64`) is Linux-only.
- Block-device size detection is implemented on Linux (`BLKGETSIZE64`) and
Windows (`IOCTL_DISK_GET_LENGTH_INFO`). On other systems only regular
files can be synced.
- If a destination path doesn't exist yet, it's created as a regular
file — clonetool won't create device nodes, so double-check device
paths for typos before running.
- SSH host keys are accepted on first connect (`StrictHostKeyChecking=accept-new`)
and rejected if they later change, same as normal SSH behavior.
- Because every run re-hashes the whole destination, a re-sync costs a
full read of both sides even when little changed — the win is in the
bytes transferred, not the bytes read. That read shows on the status
line as a `scanning destination` phase before `syncing` begins.
- Because every run re-hashes both sides in full, a re-sync costs a full
read of source and destination even when little changed — the win is in
the bytes transferred, not the bytes read. The two reads run in
parallel; the status line shows a trailing `scan H/M` until the
destination fingerprint catches up.
- `agent` is an internal subcommand spawned automatically by `sync`; it's
not meant to be run by hand, though it will work standalone for
debugging.

238
agent.go
View File

@ -75,29 +75,6 @@ func runControlAgent() error {
}
}
// readHashFrame reads frames off fr until the hash table arrives, relaying
// any "scan" progress frames the sink sends while it hashes the destination
// onward to the manager via out.
func readHashFrame(fr *FrameReader, out *FrameWriter) ([][32]byte, error) {
for {
typ, payload, err := fr.ReadFrame()
if err != nil {
return nil, fmt.Errorf("read hash table: %w", err)
}
switch typ {
case frameHashTable:
return unflattenHashes(payload)
case frameCtrlJSON:
var m CtrlMsg
if json.Unmarshal(payload, &m) == nil && m.Type == msgProgress {
_ = out.WriteJSON(m)
}
default:
return nil, fmt.Errorf("expected hash table frame, got type %d", typ)
}
}
}
func handleStat(out *FrameWriter, m CtrlMsg) {
info, err := statPath(m.Path)
if err != nil {
@ -183,16 +160,9 @@ func runPushDriver(req CtrlMsg, out *FrameWriter) {
}
// Handshake succeeded: we're committed to push for this run. The sink
// now sends the current per-block hashes of the destination it just
// read; the source loop compares against those to decide what to send.
hashes, err := readHashFrame(fr, out)
if err != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("%v (remote stderr: %s)", err, stderrBuf.String())})
return
}
// streams the destination's current per-block hashes as it scans; the
// source loop consumes them in order and reads/compares its own blocks
// as they arrive, so the two scans and the transfer all overlap.
srcFile, err := os.Open(req.Path)
if err != nil {
_ = cmd.Process.Kill()
@ -202,7 +172,7 @@ func runPushDriver(req CtrlMsg, out *FrameWriter) {
}
defer srcFile.Close()
if fatalErr := pumpPush(req, hashes, srcFile, fw, fr, out); fatalErr != nil {
if fatalErr := pumpPush(req, alignmentFor(req.Path), srcFile, fw, fr, out); fatalErr != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fatalErr.Error()})
@ -246,10 +216,15 @@ type ackEvent struct {
}
// pumpPush runs the source-side read/hash/compare/send loop against fw
// (the pipe to the remote sink) while concurrently draining ACK/ERR frames
// from fr, so it can wait for every sent block's write to be confirmed
// before declaring the push done.
func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, fr *FrameReader, out *FrameWriter) error {
// (the pipe to the remote sink) while a single reader goroutine over fr
// demultiplexes the three things the sink sends back on that one stream:
// the streamed destination block hashes (fed to the source loop as they
// arrive, so its own scan overlaps the sink's), the ACK/ERR frames for
// blocks it wrote, and any relayed scan-progress. It returns once every
// sent block's write has been confirmed.
func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *FrameReader, out *FrameWriter) error {
blockCount := (req.Size + req.BlockSize - 1) / req.BlockSize
const maxInFlight = 32
sem := make(chan struct{}, maxInFlight)
@ -265,15 +240,33 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter,
lastProgress = time.Now()
_ = out.WriteJSON(CtrlMsg{
Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped),
TotalBlocks: int64(len(hashes)),
TotalBlocks: blockCount,
})
}
// hashCh is sized to hold every block hash so the reader below never
// blocks handing hashes off (which, since ACKs share the same stream,
// would otherwise be able to deadlock against the in-flight-send limit).
// This is the same order of memory the old whole-table transfer used.
hashBuf := blockCount
if hashBuf < 1 {
hashBuf = 1
}
hashCh := make(chan blockHash, hashBuf)
ackEvents := make(chan ackEvent, 256)
go func() {
hashClosed := false
closeHash := func() {
if !hashClosed {
close(hashCh)
hashClosed = true
}
}
for {
typ, payload, err := fr.ReadFrame()
if err != nil {
closeHash()
if err == io.EOF {
ackEvents <- ackEvent{eof: true}
} else {
@ -282,18 +275,36 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter,
return
}
switch typ {
case frameBlockHash:
idx, h, derr := decodeBlockHashFrame(payload)
if derr != nil {
closeHash()
ackEvents <- ackEvent{err: derr}
return
}
hashCh <- blockHash{index: idx, hash: h}
case frameHashDone:
closeHash()
case frameAck:
idx, err := decodeIndexFrame(payload)
if err != nil {
ackEvents <- ackEvent{err: err}
idx, derr := decodeIndexFrame(payload)
if derr != nil {
closeHash()
ackEvents <- ackEvent{err: derr}
return
}
ackEvents <- ackEvent{index: idx}
case frameErr:
idx, msg, _ := decodeErrFrame(payload)
closeHash()
ackEvents <- ackEvent{err: fmt.Errorf("remote reported error at block %d: %s", idx, msg)}
return
case frameCtrlJSON:
var m CtrlMsg
if json.Unmarshal(payload, &m) == nil && m.Type == msgProgress {
_ = out.WriteJSON(m)
}
default:
closeHash()
ackEvents <- ackEvent{err: fmt.Errorf("unexpected frame type %d from sink", typ)}
return
}
@ -303,7 +314,7 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter,
sendErrCh := make(chan error, 1)
go func() {
sendErrCh <- runSourceLoop(sourceLoopParams{
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Hashes: hashes, Out: fw,
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw,
OnSkip: func(uint64) { atomic.AddInt64(&skipped, 1); maybeProgress() },
OnSend: func(idx uint64, _ [32]byte) {
sem <- struct{}{}
@ -373,7 +384,7 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter,
return fmt.Errorf("sink closed the connection with %d block write confirmation(s) still outstanding", n)
}
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), TotalBlocks: int64(len(hashes))})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), TotalBlocks: blockCount})
return nil
}
@ -421,9 +432,10 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
return
}
// Handshake succeeded. Open the destination, fingerprint its current
// content block by block, and hand that table to the source stream so it
// only sends back what differs.
// Handshake succeeded. Open the destination and fingerprint its current
// content block by block, streaming each hash to the source stream the
// moment it is computed so it can start comparing straight away; write
// whatever it streams back into the same file concurrently.
dstFile, err := os.OpenFile(req.Path, os.O_RDWR, 0)
if err != nil {
_ = cmd.Process.Kill()
@ -432,30 +444,40 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
return
}
defer dstFile.Close()
align := alignmentFor(req.Path)
hashes, err := hashFileBlocks(dstFile, req.Size, req.BlockSize, scanProgressEmitter(out))
if err != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error(), NeedPriv: isPermErr(err)})
return
}
if err := fw.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("send hash table: %v", err)})
return
}
hashErrCh := make(chan error, 1)
go func() {
err := streamHashBlocks(dstFile, req.Size, req.BlockSize, align, func(bh blockHash) error {
return fw.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out))
if err == nil {
err = fw.WriteFrame(frameHashDone, nil)
}
if err != nil {
// Unblock the source stream (waiting for more hashes) so the
// dest loop below can unwind instead of hanging.
_ = stdin.Close()
}
hashErrCh <- err
}()
loopErr := runDestLoop(destLoopParams{
File: dstFile, BlockSize: req.BlockSize, In: fr,
File: dstFile, BlockSize: req.BlockSize, Align: align, In: fr,
OnCtrlMsg: func(m CtrlMsg) {
if m.Type == msgProgress {
_ = out.WriteJSON(m)
}
},
})
hashErr := <-hashErrCh
if hashErr != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
_ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("scan destination: %v", hashErr), NeedPriv: isPermErr(hashErr)})
return
}
if loopErr != nil {
_ = cmd.Process.Kill()
_ = cmd.Wait()
@ -484,6 +506,7 @@ func runSinkRole(path string, size, blockSize int64) error {
return err
}
defer f.Close()
align := alignmentFor(path)
// Answer the handshake immediately so the push driver's short readiness
// timeout isn't spent hashing a large destination.
@ -491,23 +514,37 @@ func runSinkRole(path string, size, blockSize int64) error {
return err
}
hashes, err := hashFileBlocks(f, size, blockSize, scanProgressEmitter(out))
if err != nil {
fmt.Fprintf(os.Stderr, "sink: hash %s: %v\n", path, err)
return err
}
if err := out.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil {
return err
}
// Stream the destination's per-block hashes as they're computed while the
// dest loop below concurrently receives and writes changed blocks. A
// block is always hashed before a write for it can arrive (the source
// only sends after seeing that block's hash), so the two accesses to f
// never race on the same region.
hashErrCh := make(chan error, 1)
go func() {
err := streamHashBlocks(f, size, blockSize, align, func(bh blockHash) error {
return out.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out))
if err == nil {
err = out.WriteFrame(frameHashDone, nil)
} else {
fmt.Fprintf(os.Stderr, "sink: hash %s: %v\n", path, err)
_ = out.WriteFrame(frameErr, encodeErrFrame(0, err.Error()))
}
hashErrCh <- err
}()
return runDestLoop(destLoopParams{File: f, BlockSize: blockSize, In: in, AckOut: out})
loopErr := runDestLoop(destLoopParams{File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out})
if hashErr := <-hashErrCh; hashErr != nil {
return hashErr
}
return loopErr
}
// ---------------------------------------------------------------------
// source-stream role: one-shot process spawned (over ssh, in pull mode) on
// the source host. Reads the hash table, then performs the same
// read/hash/compare/send loop a local push driver would, writing straight
// to its own stdout.
// the source host. Consumes the streamed destination block hashes, then
// performs the same read/hash/compare/send loop a local push driver would,
// writing straight to its own stdout.
// ---------------------------------------------------------------------
func runSourceStreamRole(path string, size, blockSize int64) error {
@ -520,22 +557,47 @@ func runSourceStreamRole(path string, size, blockSize int64) error {
return err
}
defer f.Close()
align := alignmentFor(path)
if err := out.WriteFrame(frameReady, nil); err != nil {
return err
}
typ, payload, err := in.ReadFrame()
if err != nil {
return fmt.Errorf("source-stream: read hash table: %w", err)
}
if typ != frameHashTable {
return fmt.Errorf("source-stream: expected hash table frame, got type %d", typ)
}
hashes, err := unflattenHashes(payload)
if err != nil {
return err
blockCount := (size + blockSize - 1) / blockSize
hashBuf := blockCount
if hashBuf < 1 {
hashBuf = 1
}
hashCh := make(chan blockHash, hashBuf)
readErrCh := make(chan error, 1)
go func() {
for {
typ, payload, err := in.ReadFrame()
if err != nil {
readErrCh <- fmt.Errorf("source-stream: read hash stream: %w", err)
close(hashCh)
return
}
switch typ {
case frameBlockHash:
idx, h, derr := decodeBlockHashFrame(payload)
if derr != nil {
readErrCh <- derr
close(hashCh)
return
}
hashCh <- blockHash{index: idx, hash: h}
case frameHashDone:
readErrCh <- nil
close(hashCh)
return
default:
readErrCh <- fmt.Errorf("source-stream: unexpected frame type %d", typ)
close(hashCh)
return
}
}
}()
var copied, skipped int64
var lastProgress time.Time
@ -544,12 +606,16 @@ func runSourceStreamRole(path string, size, blockSize int64) error {
return
}
lastProgress = time.Now()
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: copied, Skipped: skipped, TotalBlocks: int64(len(hashes))})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: copied, Skipped: skipped, TotalBlocks: blockCount})
}
return runSourceLoop(sourceLoopParams{
File: f, Size: size, BlockSize: blockSize, Hashes: hashes, Out: out,
loopErr := runSourceLoop(sourceLoopParams{
File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
OnSkip: func(uint64) { skipped++; maybeProgress() },
OnSend: func(uint64, [32]byte) { copied++; maybeProgress() },
})
if readErr := <-readErrCh; readErr != nil {
return readErr
}
return loopErr
}

21
build.sh Executable file
View File

@ -0,0 +1,21 @@
#!/bin/sh
# Build static clonetool binaries for the common OS/arch targets into dist/.
set -e
cd "$(dirname "$0")"
mkdir -p dist
build() {
os=$1 arch=$2 out=$3
echo "==> $os/$arch -> dist/$out"
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath -ldflags "-s -w" -o "dist/$out" .
}
build linux amd64 clonetool-linux-amd64
build linux arm64 clonetool-linux-arm64
build windows amd64 clonetool-windows-amd64.exe
build windows arm64 clonetool-windows-arm64.exe
build darwin amd64 clonetool-darwin-amd64
build darwin arm64 clonetool-darwin-arm64
echo "done:"
ls -la dist/

19
control.go
View File

@ -197,22 +197,3 @@ func (c *Controller) Close() error {
}
return nil
}
func flattenHashes(hashes [][32]byte) []byte {
buf := make([]byte, len(hashes)*32)
for i, h := range hashes {
copy(buf[i*32:], h[:])
}
return buf
}
func unflattenHashes(b []byte) ([][32]byte, error) {
if len(b)%32 != 0 {
return nil, fmt.Errorf("hash table: %d bytes is not a multiple of 32", len(b))
}
out := make([][32]byte, len(b)/32)
for i := range out {
copy(out[i][:], b[i*32:i*32+32])
}
return out, nil
}

5
ctrlmsg.go
View File

@ -39,8 +39,11 @@ type CtrlMsg struct {
Reason string `json:"reason,omitempty"`
Message string `json:"message,omitempty"`
// progress
// progress. Scan and transfer now overlap, so a run emits both "scan"
// messages (Hashed/TotalBlocks) and transfer messages (Copied/Skipped)
// interleaved; the printer keeps the latest of each.
Phase string `json:"phase,omitempty"` // "scan" while hashing the destination; empty during transfer
Hashed int64 `json:"hashed,omitempty"`
Copied int64 `json:"copied,omitempty"`
Skipped int64 `json:"skipped,omitempty"`
TotalBlocks int64 `json:"totalBlocks,omitempty"`

16
device.go
View File

@ -14,6 +14,19 @@ type PathInfo struct {
}
func statPath(path string) (PathInfo, error) {
// A raw disk handle (Windows \\.\PhysicalDrive0, \\.\C:) is not something
// os.Stat can describe, so ask the platform for its size directly.
if isDevicePath(path) {
size, err := blockDeviceSize(path)
if err != nil {
if os.IsNotExist(err) {
return PathInfo{Exists: false}, nil
}
return PathInfo{}, fmt.Errorf("stat device %s: %w", path, err)
}
return PathInfo{Exists: true, IsDevice: true, Size: size}, nil
}
fi, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
@ -50,6 +63,9 @@ func prepareDest(path string, targetSize int64) error {
}
return nil
}
if isDevicePath(path) && !info.Exists {
return fmt.Errorf("destination device %s not found", path)
}
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return fmt.Errorf("open/create %s: %w", path, err)

4
device_other.go
View File

@ -1,9 +1,9 @@
//go:build !linux
//go:build !linux && !windows
package main
import "fmt"
func blockDeviceSize(path string) (int64, error) {
return 0, fmt.Errorf("block device size detection is only implemented on linux (got path %s)", path)
return 0, fmt.Errorf("block device size detection is only implemented on linux and windows (got path %s)", path)
}

18
device_unix.go Normal file
View File

@ -0,0 +1,18 @@
//go:build !windows
package main
// isDevicePath reports whether a path must be treated as a raw device from
// its string form alone. On unix, os.Stat's mode bits already identify
// device nodes accurately (symlinks, non-standard locations and all), so
// this is always false and statPath relies on those instead.
func isDevicePath(string) bool { return false }
// alignmentFor returns the offset/length alignment a path's handle requires
// for positioned reads and writes. Unix block devices accept ordinary
// buffered pread/pwrite at any alignment, so there is nothing to round to.
func alignmentFor(string) int64 { return 1 }
// canElevate reports whether a permission failure opening a device is worth
// retrying under `sudo` (see --sudo). Always true on unix.
func canElevate() bool { return true }

99
device_windows.go Normal file
View File

@ -0,0 +1,99 @@
//go:build windows
package main
import (
"fmt"
"os"
"strings"
"syscall"
"unsafe"
)
// isDevicePath recognises the Win32 device namespaces clonetool can sync
// against by string form: \\.\PhysicalDrive0 (a whole disk) and \\.\C: (a
// volume). Both the \\.\ and \\?\ prefixes, and their forward-slash
// spellings, are accepted. os.Stat can't describe these paths, so statPath
// keys off this instead of mode bits on Windows.
func isDevicePath(path string) bool {
p := strings.ReplaceAll(path, "/", `\`)
return strings.HasPrefix(p, `\\.\`) || strings.HasPrefix(p, `\\?\`)
}
// canElevate is false on Windows: there is no `sudo` equivalent to re-exec
// under, so a permission failure opening a raw disk is reported with a hint
// to run from an elevated console instead of being retried.
func canElevate() bool { return false }
const (
ioctlDiskGetLengthInfo = 0x0007405C // IOCTL_DISK_GET_LENGTH_INFO
ioctlDiskGetDriveGeometry = 0x00070000 // IOCTL_DISK_GET_DRIVE_GEOMETRY
)
// diskGeometry mirrors DISK_GEOMETRY (24 bytes; Cylinders is a LARGE_INTEGER).
type diskGeometry struct {
Cylinders int64
MediaType uint32
TracksPerCylinder uint32
SectorsPerTrack uint32
BytesPerSector uint32
}
// openDeviceHandle opens path for a metadata ioctl only: zero access rights
// (which need no privilege and don't require the volume to be unlocked) and
// shared read/write so it doesn't disturb a mounted filesystem.
func openDeviceHandle(path string) (syscall.Handle, error) {
p, err := syscall.UTF16PtrFromString(path)
if err != nil {
return syscall.InvalidHandle, err
}
return syscall.CreateFile(p, 0,
syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE, nil,
syscall.OPEN_EXISTING, 0, 0)
}
// blockDeviceSize returns the byte length of a physical drive or volume via
// DeviceIoControl(IOCTL_DISK_GET_LENGTH_INFO).
func blockDeviceSize(path string) (int64, error) {
h, err := openDeviceHandle(path)
if err != nil {
return 0, &os.PathError{Op: "open", Path: path, Err: err}
}
defer syscall.CloseHandle(h)
var length int64 // GET_LENGTH_INFORMATION is a single LARGE_INTEGER
var ret uint32
err = syscall.DeviceIoControl(h, ioctlDiskGetLengthInfo,
nil, 0,
(*byte)(unsafe.Pointer(&length)), uint32(unsafe.Sizeof(length)),
&ret, nil)
if err != nil {
return 0, fmt.Errorf("IOCTL_DISK_GET_LENGTH_INFO %s: %w", path, err)
}
return length, nil
}
// alignmentFor returns the sector size a raw disk handle's positioned reads
// and writes must be aligned to; 1 for an ordinary file path. It falls back
// to 512 if the geometry query fails.
func alignmentFor(path string) int64 {
if !isDevicePath(path) {
return 1
}
h, err := openDeviceHandle(path)
if err != nil {
return 512
}
defer syscall.CloseHandle(h)
var g diskGeometry
var ret uint32
err = syscall.DeviceIoControl(h, ioctlDiskGetDriveGeometry,
nil, 0,
(*byte)(unsafe.Pointer(&g)), uint32(unsafe.Sizeof(g)),
&ret, nil)
if err != nil || g.BytesPerSector == 0 {
return 512
}
return int64(g.BytesPerSector)
}

19
frame.go
View File

@ -22,12 +22,13 @@ type frameType byte
const (
frameCtrlJSON frameType = 0x01 // payload: JSON object (control channel)
frameHashTable frameType = 0x02 // payload: raw concatenated 32-byte hashes
frameBlockHash frameType = 0x02 // payload: 8-byte index + 32-byte hash (one destination block, streamed in index order)
frameData frameType = 0x03 // payload: 8-byte index + 32-byte hash + block bytes
frameAck frameType = 0x04 // payload: 8-byte index
frameErr frameType = 0x05 // payload: 8-byte index + UTF-8 message
frameDone frameType = 0x06 // payload: empty
frameReady frameType = 0x07 // payload: empty
frameHashDone frameType = 0x08 // payload: empty — end of the frameBlockHash stream
)
// FrameWriter serializes concurrent writers onto one underlying stream.
@ -108,6 +109,22 @@ func decodeDataFrame(b []byte) (index uint64, hash [32]byte, payload []byte, err
return index, hash, payload, nil
}
func encodeBlockHashFrame(index uint64, hash [32]byte) []byte {
buf := make([]byte, 8+32)
binary.BigEndian.PutUint64(buf[0:8], index)
copy(buf[8:40], hash[:])
return buf
}
func decodeBlockHashFrame(b []byte) (index uint64, hash [32]byte, err error) {
if len(b) < 40 {
return 0, hash, fmt.Errorf("block-hash frame too short: %d bytes", len(b))
}
index = binary.BigEndian.Uint64(b[0:8])
copy(hash[:], b[8:40])
return index, hash, nil
}
func encodeIndexFrame(index uint64) []byte {
buf := make([]byte, 8)
binary.BigEndian.PutUint64(buf, index)

65
manager.go
View File

@ -144,7 +144,7 @@ func runSync(cfg SyncConfig) error {
// restart under sudo. The returned bool reports whether the agent (and any
// peer helper it later spawns for this side) is running elevated.
func bringUpController(spec Spec, tag string, cfg *SyncConfig, remoteBin, probePath string) (*Controller, PathInfo, bool, error) {
sudo := cfg.Sudo == "always"
sudo := cfg.Sudo == "always" && canElevate()
c, err := startController(spec, tag, cfg, remoteBin, sudo)
if err != nil {
return nil, PathInfo{}, sudo, err
@ -153,6 +153,11 @@ func bringUpController(spec Spec, tag string, cfg *SyncConfig, remoteBin, probeP
if err == nil {
return c, info, sudo, nil
}
if errors.Is(err, errNeedPriv) && !canElevate() {
c.Close()
return nil, PathInfo{}, sudo, fmt.Errorf(
"%w; on Windows, run clonetool from an elevated (Administrator) console to open a raw disk", err)
}
if errors.Is(err, errNeedPriv) && cfg.Sudo == "auto" && !sudo {
fmt.Fprintf(os.Stderr, "%s: permission denied on %s; retrying via sudo ...\n", tag, probePath)
c.Close()
@ -229,9 +234,11 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool {
return line == "y" || line == "yes"
}
// progressPrinter renders a single rewriting status line on stderr for both
// phases of a run: "scan" while a side hashes the destination, then the
// block transfer itself (with a rolling copy rate).
// progressPrinter renders a single rewriting status line on stderr. Scan
// and transfer now run concurrently, so it keeps the latest of each kind of
// update (blocks hashed so far, blocks copied/skipped) and shows them on one
// line, with a rolling copy rate; the "scan H/M" segment disappears once the
// destination fingerprint is complete.
type progressPrinter struct {
blockSize int64
start time.Time
@ -239,6 +246,11 @@ type progressPrinter struct {
lastCopied int64
rate float64 // copied blocks/sec, smoothed
active bool
hashed int64
copied int64
skipped int64
total int64
}
func newProgressPrinter(blockSize int64) *progressPrinter {
@ -248,34 +260,41 @@ func newProgressPrinter(blockSize int64) *progressPrinter {
func (p *progressPrinter) print(m CtrlMsg) {
p.active = true
if m.TotalBlocks > 0 {
p.total = m.TotalBlocks
}
if m.Phase == "scan" {
fmt.Fprintf(os.Stderr, "\r scanning destination %s %s/%s blocks ",
progressBar(m.Copied, m.TotalBlocks),
formatCount(m.Copied), formatCount(m.TotalBlocks))
return
if m.Hashed > p.hashed {
p.hashed = m.Hashed
}
} else {
p.copied, p.skipped = m.Copied, m.Skipped
now := time.Now()
if dt := now.Sub(p.lastT).Seconds(); dt >= 0.4 {
inst := float64(m.Copied-p.lastCopied) / dt
if p.rate == 0 {
p.rate = inst
} else {
p.rate = 0.6*p.rate + 0.4*inst
}
p.lastT, p.lastCopied = now, m.Copied
}
}
done := m.Copied + m.Skipped
total := m.TotalBlocks
done := p.copied + p.skipped
total := p.total
if total <= 0 {
total = done
}
now := time.Now()
if dt := now.Sub(p.lastT).Seconds(); dt >= 0.4 {
inst := float64(m.Copied-p.lastCopied) / dt
if p.rate == 0 {
p.rate = inst
} else {
p.rate = 0.6*p.rate + 0.4*inst
}
p.lastT, p.lastCopied = now, m.Copied
}
mib := p.rate * float64(p.blockSize) / (1024 * 1024)
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s %6.1f MiB/s ",
scan := ""
if p.total > 0 && p.hashed < p.total {
scan = fmt.Sprintf(" scan %s/%s", formatCount(p.hashed), formatCount(p.total))
}
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s %6.1f MiB/s ",
progressBar(done, total), formatCount(done), formatCount(total),
formatCount(m.Copied), formatCount(m.Skipped), mib)
formatCount(p.copied), formatCount(p.skipped), scan, mib)
}
// finish ends the current status line with a newline so following output

141
syncside.go
View File

@ -9,10 +9,13 @@ import (
"time"
)
// readBlockAt reads exactly min(len(buf), totalSize-offset) bytes at offset,
// treating a fully-satisfied read as success even if the underlying
// implementation also reports io.EOF for it.
func readBlockAt(f *os.File, buf []byte, offset, totalSize int64) (int, error) {
// readBlockAt reads exactly min(len(buf), totalSize-offset) logical bytes at
// offset, treating a fully-satisfied read as success even if the underlying
// implementation also reports io.EOF for it. When align > 1 (a raw disk
// handle that only accepts sector-aligned I/O) the physical read is rounded
// up to the next multiple of align, but the returned count is still the
// logical size — the caller only ever looks at buf[:n].
func readBlockAt(f *os.File, buf []byte, offset, totalSize, align int64) (int, error) {
remaining := totalSize - offset
if remaining <= 0 {
return 0, io.EOF
@ -21,40 +24,94 @@ func readBlockAt(f *os.File, buf []byte, offset, totalSize int64) (int, error) {
if remaining < want {
want = remaining
}
n, err := f.ReadAt(buf[:want], offset)
if err != nil && !(err == io.EOF && int64(n) == want) {
readLen := want
if align > 1 && want%align != 0 {
readLen = roundUp(want, align)
if readLen > int64(len(buf)) {
readLen = int64(len(buf))
}
}
n, err := f.ReadAt(buf[:readLen], offset)
if err != nil && !(err == io.EOF && int64(n) >= want) {
return n, err
}
if int64(n) > want {
n = int(want)
}
return n, nil
}
// hashFileBlocks reads f in blockSize-byte blocks up to size and returns the
// SHA-256 of each. This is how a destination fingerprints its *current*
// content at the start of every sync — clonetool keeps no hash state of its
// own between runs. onProgress, if non-nil, is called with (blocksHashed,
// totalBlocks) before the first block and after each one.
func hashFileBlocks(f *os.File, size, blockSize int64, onProgress func(done, total int64)) ([][32]byte, error) {
// writeBlockAt writes block at offset. When align > 1 and the block is not a
// whole number of sectors (only ever the final block of a device sync), the
// enclosing aligned span is read first and the block overlaid onto it, so a
// raw disk handle that rejects sub-sector writes still gets an aligned write
// and the bytes past the sync size are preserved.
func writeBlockAt(f *os.File, block []byte, offset, align int64) error {
if align <= 1 || int64(len(block))%align == 0 {
_, err := f.WriteAt(block, offset)
return err
}
padded := roundUp(int64(len(block)), align)
tmp := make([]byte, padded)
if _, err := f.ReadAt(tmp, offset); err != nil && err != io.EOF {
return fmt.Errorf("read-modify-write tail at %d: %w", offset, err)
}
copy(tmp, block)
_, err := f.WriteAt(tmp, offset)
return err
}
// checkBlockAlign rejects a block size that a raw disk handle's sector
// alignment can't satisfy (every block offset is a multiple of the block
// size, so the block size itself must be a whole number of sectors).
func checkBlockAlign(blockSize, align int64) error {
if align > 1 && blockSize%align != 0 {
return fmt.Errorf("block size %d is not a multiple of the device's %d-byte sector size; pass --block-size divisible by %d",
blockSize, align, align)
}
return nil
}
// blockHash pairs a block index with the SHA-256 of the destination's
// current content for that block. The destination side streams these in
// ascending index order as it scans; the source side consumes them in the
// same order, so the two scans overlap instead of running back to back.
type blockHash struct {
index uint64
hash [32]byte
}
// streamHashBlocks reads f in blockSize-byte blocks up to size, hashing each
// one and handing it to onHash the moment it is computed (this is what lets
// the peer start comparing before the whole side has been scanned). It keeps
// no hash state of its own. onProgress, if non-nil, is called with
// (blocksHashed, totalBlocks) before the first block and after each one.
func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh blockHash) error, onProgress func(done, total int64)) error {
if err := checkBlockAlign(blockSize, align); err != nil {
return err
}
blockCount := (size + blockSize - 1) / blockSize
out := make([][32]byte, blockCount)
buf := make([]byte, blockSize)
if onProgress != nil {
onProgress(0, blockCount)
}
for i := int64(0); i < blockCount; i++ {
n, err := readBlockAt(f, buf, i*blockSize, size)
n, err := readBlockAt(f, buf, i*blockSize, size, align)
if err != nil {
return nil, fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err)
return fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err)
}
if err := onHash(blockHash{index: uint64(i), hash: sha256.Sum256(buf[:n])}); err != nil {
return err
}
out[i] = sha256.Sum256(buf[:n])
if onProgress != nil {
onProgress(i+1, blockCount)
}
}
return out, nil
return nil
}
// scanProgressEmitter returns an onProgress callback for hashFileBlocks that
// forwards "scan" phase progress onto out, throttled to ~3/second.
// scanProgressEmitter returns an onProgress callback for streamHashBlocks
// that forwards "scan" phase progress onto out, throttled to ~3/second.
func scanProgressEmitter(out *FrameWriter) func(done, total int64) {
var last time.Time
return func(done, total int64) {
@ -62,46 +119,55 @@ func scanProgressEmitter(out *FrameWriter) func(done, total int64) {
return
}
last = time.Now()
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "scan", Copied: done, TotalBlocks: total})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "scan", Hashed: done, TotalBlocks: total})
}
}
// sourceLoopParams drives the single read -> hash -> compare -> maybe-send
// pass over a source path. It is used both by the standalone
// "source-stream" role (writing to its own stdout) and by a source control
// agent's push driver (writing into a spawned ssh subprocess's stdin).
// pass over a source path. The hash of each destination block arrives on
// Hashes (in ascending index order); the block read for the comparison is
// the same buffer that gets sent on if it differs — a differing block is
// never read twice.
type sourceLoopParams struct {
File *os.File // already opened for reading by the caller
Size int64
BlockSize int64
Hashes [][32]byte // previous known hashes, len == block count
Align int64
Hashes <-chan blockHash
Out *FrameWriter
OnSkip func(index uint64)
OnSend func(index uint64, hash [32]byte) // called after the DATA frame is written
// OnSend is called for a differing block immediately before its DATA
// frame goes on the wire — never after. The push driver relies on this
// ordering to record the block as awaiting confirmation before the peer
// can possibly ACK it (over a fast local pipe the ACK really can arrive
// first), and to apply in-flight backpressure before the send.
OnSend func(index uint64, hash [32]byte)
}
func runSourceLoop(p sourceLoopParams) error {
if err := checkBlockAlign(p.BlockSize, p.Align); err != nil {
return err
}
f := p.File
buf := make([]byte, p.BlockSize)
blockCount := uint64(len(p.Hashes))
for index := uint64(0); index < blockCount; index++ {
offset := int64(index) * p.BlockSize
n, err := readBlockAt(f, buf, offset, p.Size)
for bh := range p.Hashes {
offset := int64(bh.index) * p.BlockSize
n, err := readBlockAt(f, buf, offset, p.Size, p.Align)
if err != nil {
return fmt.Errorf("read %s at block %d: %w", f.Name(), index, err)
return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err)
}
hash := sha256.Sum256(buf[:n])
if hash == p.Hashes[index] {
if hash == bh.hash {
if p.OnSkip != nil {
p.OnSkip(index)
p.OnSkip(bh.index)
}
continue
}
if err := p.Out.WriteFrame(frameData, encodeDataFrame(index, hash, buf[:n])); err != nil {
return fmt.Errorf("send block %d: %w", index, err)
}
if p.OnSend != nil {
p.OnSend(index, hash)
p.OnSend(bh.index, hash)
}
if err := p.Out.WriteFrame(frameData, encodeDataFrame(bh.index, hash, buf[:n])); err != nil {
return fmt.Errorf("send block %d: %w", bh.index, err)
}
}
return p.Out.WriteFrame(frameDone, nil)
@ -116,6 +182,7 @@ func runSourceLoop(p sourceLoopParams) error {
type destLoopParams struct {
File *os.File // already opened for read/write by the caller
BlockSize int64
Align int64 // sector alignment for a raw disk handle, else 0/1
In *FrameReader
AckOut *FrameWriter // optional
// OnCtrlMsg handles an interleaved frameCtrlJSON frame (used only in
@ -145,7 +212,7 @@ func runDestLoop(p destLoopParams) error {
}
return fmt.Errorf("block %d: hash mismatch after transfer", index)
}
if _, err := f.WriteAt(block, int64(index)*p.BlockSize); err != nil {
if err := writeBlockAt(f, block, int64(index)*p.BlockSize, p.Align); err != nil {
if p.AckOut != nil {
_ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, err.Error()))
}

8
util.go
View File

@ -34,6 +34,14 @@ func (b *limitedBuffer) String() string {
return string(b.buf)
}
// roundUp rounds n up to the next multiple of to (to <= 1 is a no-op).
func roundUp(n, to int64) int64 {
if to <= 1 {
return n
}
return (n + to - 1) / to * to
}
func humanBytes(n int64) string {
const unit = 1024
if n < unit {