diff --git a/.gitignore b/.gitignore index 6490ecf..672fb67 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /clonetool +/dist/ diff --git a/README.md b/README.md index fd7da02..824b68b 100644 --- a/README.md +++ b/README.md @@ -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` (whole disk) or `\\.\:` (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. diff --git a/agent.go b/agent.go index 668bb2a..5f4d451 100644 --- a/agent.go +++ b/agent.go @@ -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 } diff --git a/build.sh b/build.sh new file mode 100755 index 0000000..8585b1d --- /dev/null +++ b/build.sh @@ -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/ diff --git a/control.go b/control.go index 61d1dfa..30eede4 100644 --- a/control.go +++ b/control.go @@ -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 -} diff --git a/ctrlmsg.go b/ctrlmsg.go index 0553082..6ed4ccb 100644 --- a/ctrlmsg.go +++ b/ctrlmsg.go @@ -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"` diff --git a/device.go b/device.go index 068cdd0..6a9a0c2 100644 --- a/device.go +++ b/device.go @@ -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) diff --git a/device_other.go b/device_other.go index ae13f7e..0370ed7 100644 --- a/device_other.go +++ b/device_other.go @@ -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) } diff --git a/device_unix.go b/device_unix.go new file mode 100644 index 0000000..11e9aa4 --- /dev/null +++ b/device_unix.go @@ -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 } diff --git a/device_windows.go b/device_windows.go new file mode 100644 index 0000000..abd4969 --- /dev/null +++ b/device_windows.go @@ -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) +} diff --git a/frame.go b/frame.go index fcef468..7579d2c 100644 --- a/frame.go +++ b/frame.go @@ -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) diff --git a/manager.go b/manager.go index 28683e7..ac05113 100644 --- a/manager.go +++ b/manager.go @@ -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 diff --git a/syncside.go b/syncside.go index a898a3b..f3b2c9c 100644 --- a/syncside.go +++ b/syncside.go @@ -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())) } diff --git a/util.go b/util.go index 643bbed..dd8b3f9 100644 --- a/util.go +++ b/util.go @@ -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 {