Compare commits

...

8 Commits
v0.1 ... main

14 changed files with 359 additions and 281 deletions

117
README.md
View File

@ -1,9 +1,27 @@
# WARNING
THIS TOOL IS COMPLETELY VIBE-CODED!!!
THESE LINES ARE THE ONLY ONE I HAVE WRITTEN MYSELF!!!!
I SWEAR I DID NOT READ MORE THAN 20 LINES OF CODE!
USE AT YOUR OWN RISK (AS ALWAYS)
I needed a Tool to sync 2 Files from a to b with differential transfer, all managed from a workstation c.
Maybe i use this to learn go...
Looks nice, builds fast, runs everywhere with one Binary.
# clonetool
Block-level sync for a large file or block device between two machines (or
two paths on the same machine), driven from a third, passive "manager"
machine. Single static Go binary, no runtime dependencies beyond the
system `ssh` client for remote endpoints.
system `ssh` client for remote endpoints. Linux only.
## How it works
@ -44,13 +62,25 @@ system `ssh` client for remote endpoints.
is used at all — the source agent spawns the write-side helper as a
plain local subprocess.
- **Self-deploy:** if a remote endpoint has no runnable `clonetool` on
`PATH` (or wherever `--remote-bin` points), the manager streams *this*
binary to `~/.clonetool/bin/clonetool` on that host over the existing
SSH connection and uses it — no install, no root. Disable with
`--deploy=false`. If the copied binary won't execute there (wrong CPU
architecture) the error says so; build one for the remote's arch
`PATH` (or wherever `--remote-bin` points), **or the one that's there is
a different build than this binary** (see below), the manager streams
*this* binary to `~/.clonetool/bin/clonetool` on that host over the
existing SSH connection and uses it — no install, no root. Disable with
`--deploy=false` (a stale or missing binary is then an error, except a
merely-stale `--remote-bin` is used anyway with a warning, since there's
no way to correct it). If the copied binary won't execute there (wrong
CPU architecture) the error says so; build one for the remote's arch
(`CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build`) and put it on `PATH`
or pass `--remote-bin`.
- **Version/staleness check:** every binary has a build timestamp baked in
(see Build below) and `clonetool version` prints it. Before using a
remote `clonetool`, the manager runs its `version` there and compares
that timestamp against its own — a mismatch (e.g. the remote was
self-deployed from an older build of this tool) is treated the same as
"not runnable" and triggers the self-deploy above, so the remote binary
is kept in sync with whatever you're running locally. Only the build
timestamp is compared, not `GOARCH` — that is expected to differ across
a cross-compiled deploy.
- **sudo:** if reading or writing an endpoint that is a **block device**
fails with a permission error, `--sudo=auto` (the default) transparently
restarts that side's agent — and the helper it spawns on the peer —
@ -61,8 +91,8 @@ system `ssh` client for remote endpoints.
start; `--sudo=never` never does.
- Sizing rules:
- Destination is a **block device**: it can't be resized, so if the
source is larger the job fails; otherwise exactly `min(source, dest)`
bytes are synced and the remainder of the device is left untouched.
source is larger the job fails; otherwise exactly the source's size is
synced and the remainder of the device is left untouched.
- Destination is a **regular file**: it's truncated (created if
missing) to exactly the source's size, growing or shrinking it.
Shrinking an existing non-empty file prompts for confirmation unless
@ -74,54 +104,27 @@ 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):
Cross-compile for another architecture by setting `GOARCH` (no cgo, so
this works 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 .
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o clonetool .
```
`./build.sh` writes all three of the above (plus linux/amd64) into `dist/`.
`./build.sh` writes linux/amd64 and linux/arm64 binaries into `dist/`. It
also stamps every binary it builds with the same build timestamp (via
`-ldflags -X main.buildTime=...`), which is how the manager tells a stale
self-deployed remote binary apart from a current one (see "Version/staleness
check" above) — build all your binaries for a given release with one
`./build.sh` run so they compare equal to each other. A plain `go build`
with no `-ldflags` leaves the timestamp at `"dev"`, which still works for
this check (it only ever compares equal to another literal copy of the same
`dev` binary, never to a real timestamp).
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 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`.
cross-compile if your hosts differ.
## Usage
@ -136,6 +139,9 @@ clonetool sync --source LOC --dest LOC [options]
# Same machine
clonetool sync --source /dev/sda --dest /dev/sdb
# Whole disk to an image file
clonetool sync --source /dev/sda --dest /srv/sda.img
# Two remote machines, orchestrated from a third
clonetool sync --source db1:/dev/vdb --dest backup-host:/srv/db1.img
@ -158,14 +164,19 @@ Options:
| `--remote-bin` | `clonetool` | Path to clonetool on remote hosts. |
| `--manager-host` | local hostname | Address a peer should use to reach this machine, needed only when source or dest is local to the manager *and* the other side is remote and ends up needing to dial back in (pull fallback). |
`clonetool version` prints the binary's `GOOS/GOARCH` (used internally for
the self-deploy check).
`clonetool version` prints the binary's `GOOS/GOARCH` and build timestamp,
e.g. `clonetool linux/amd64 build=2024-06-01T12:00:00Z` (the timestamp is
used internally for the self-deploy staleness check above).
The status line shows read throughput for both sides and write throughput
for the destination separately: `rd(src)` is the source reading and
comparing its blocks, `rd(dst)` is the destination fingerprinting its
current content (only while the trailing `scan H/M` segment is present),
and `wr(dst)` is the destination actually writing changed blocks.
## Caveats
- 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.
- Block-device size detection uses `BLKGETSIZE64`; the tool is Linux only.
- 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.

113
agent.go
View File

@ -19,6 +19,7 @@ func cmdAgent(args []string) error {
role := fs.String("role", "", "control|sink|source-stream (internal)")
path := fs.String("path", "", "path to read/write")
size := fs.Int64("size", 0, "total sync size in bytes")
base := fs.Int64("base", 0, "byte offset the window starts at")
blockSize := fs.Int64("block-size", defaultBlockSize, "block size in bytes")
if err := fs.Parse(args); err != nil {
return err
@ -28,11 +29,11 @@ func cmdAgent(args []string) error {
case "control":
return runControlAgent()
case "sink":
return runSinkRole(*path, *size, *blockSize)
return runSinkRole(*path, *base, *size, *blockSize)
case "source-stream":
return runSourceStreamRole(*path, *size, *blockSize)
return runSourceStreamRole(*path, *base, *size, *blockSize)
default:
return fmt.Errorf("agent: unknown or missing --role %q (want control|sink|source-stream)", *role)
return fmt.Errorf("agent: unknown or missing --role %q", *role)
}
}
@ -127,6 +128,7 @@ func runPushDriver(req CtrlMsg, out *FrameWriter) {
tailArgs := []string{
"agent", "--role", "sink",
"--path", req.PeerPath,
"--base", strconv.FormatInt(req.PeerBase, 10),
"--size", strconv.FormatInt(req.Size, 10),
"--block-size", strconv.FormatInt(req.BlockSize, 10),
}
@ -231,7 +233,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
var pendingMu sync.Mutex
pending := make(map[uint64]bool)
var copied, skipped int64
var copied, skipped, srcRead int64
var lastProgress time.Time
maybeProgress := func() {
if time.Since(lastProgress) < 500*time.Millisecond {
@ -239,11 +241,24 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
}
lastProgress = time.Now()
_ = out.WriteJSON(CtrlMsg{
Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped),
Type: msgProgress, Phase: "xfer",
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
TotalBlocks: blockCount,
})
}
// lastActivityNano records the last moment either side made progress: a
// frame arriving from the sink (hashes, ACKs, relayed progress) or the
// source loop reading and comparing a block. The idle watchdog below
// treats a stall as fatal only when *neither* side has moved for a whole
// window — otherwise a check-only run, where every block matches and so no
// DATA frames and no ACKs are ever exchanged, trips the watchdog purely
// because the wire went quiet while the two scans churned through
// identical data.
var lastActivityNano int64
atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano())
markActivity := func() { atomic.StoreInt64(&lastActivityNano, time.Now().UnixNano()) }
// 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).
@ -274,6 +289,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
}
return
}
markActivity()
switch typ {
case frameBlockHash:
idx, h, derr := decodeBlockHashFrame(payload)
@ -315,8 +331,10 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
go func() {
sendErrCh <- runSourceLoop(sourceLoopParams{
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw,
OnSkip: func(uint64) { atomic.AddInt64(&skipped, 1); maybeProgress() },
OnRead: func(uint64) { markActivity(); atomic.AddInt64(&srcRead, 1); maybeProgress() },
OnSkip: func(uint64) { markActivity(); atomic.AddInt64(&skipped, 1); maybeProgress() },
OnSend: func(idx uint64, _ [32]byte) {
markActivity()
sem <- struct{}{}
pendingMu.Lock()
pending[idx] = true
@ -332,7 +350,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
sendCh := sendErrCh
ackCh := ackEvents
var fatalErr error
idle := time.NewTimer(120 * time.Second)
// idleTimeout is how long *both* sides can be silent before the transfer
// is declared stalled. The timer may fire sooner; a fire is only fatal
// when lastActivityNano confirms nothing has moved for the whole window.
const idleTimeout = 120 * time.Second
idle := time.NewTimer(idleTimeout)
defer idle.Stop()
for sendCh != nil || ackCh != nil {
if !idle.Stop() {
@ -341,7 +363,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
default:
}
}
idle.Reset(120 * time.Second)
idle.Reset(idleTimeout)
select {
case sendErr := <-sendCh:
sendCh = nil
@ -367,7 +389,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
maybeProgress()
}
case <-idle.C:
fatalErr = fmt.Errorf("timed out waiting for the sink")
idleFor := time.Since(time.Unix(0, atomic.LoadInt64(&lastActivityNano)))
if idleFor < idleTimeout {
continue // progress somewhere within the window; re-arm and wait
}
fatalErr = fmt.Errorf("timed out waiting for the sink (no activity for %s)", idleFor.Round(time.Second))
}
if fatalErr != nil {
break
@ -384,7 +410,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
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: blockCount})
_ = out.WriteJSON(CtrlMsg{
Type: msgProgress, Phase: "xfer",
Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), SrcRead: atomic.LoadInt64(&srcRead),
TotalBlocks: blockCount,
})
return nil
}
@ -400,6 +430,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
tailArgs := []string{
"agent", "--role", "source-stream",
"--path", req.PeerPath,
"--base", strconv.FormatInt(req.PeerBase, 10),
"--size", strconv.FormatInt(req.Size, 10),
"--block-size", strconv.FormatInt(req.BlockSize, 10),
}
@ -448,7 +479,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
hashErrCh := make(chan error, 1)
go func() {
err := streamHashBlocks(dstFile, req.Size, req.BlockSize, align, func(bh blockHash) error {
err := streamHashBlocks(dstFile, req.Base, req.Size, req.BlockSize, align, func(bh blockHash) error {
return fw.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out))
if err == nil {
@ -462,14 +493,24 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
hashErrCh <- err
}()
var written int64
var lastWritten time.Time
loopErr := runDestLoop(destLoopParams{
File: dstFile, BlockSize: req.BlockSize, Align: align, In: fr,
File: dstFile, Base: req.Base, BlockSize: req.BlockSize, Align: align, In: fr,
OnCtrlMsg: func(m CtrlMsg) {
if m.Type == msgProgress {
_ = out.WriteJSON(m)
}
},
OnWrite: func(uint64) {
written++
if time.Since(lastWritten) >= 500*time.Millisecond {
lastWritten = time.Now()
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
}
},
})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
hashErr := <-hashErrCh
if hashErr != nil {
@ -496,7 +537,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
// destination host. Dumb write endpoint: verify+pwrite+ack per block.
// ---------------------------------------------------------------------
func runSinkRole(path string, size, blockSize int64) error {
func runSinkRole(path string, base, size, blockSize int64) error {
out := NewFrameWriter(os.Stdout)
in := NewFrameReader(os.Stdin)
@ -519,11 +560,33 @@ func runSinkRole(path string, size, blockSize int64) error {
// 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.
// While the destination scan runs, emit a heartbeat on a fixed cadence
// even if streamHashBlocks is parked inside one slow ReadAt (a big /
// non-sparse / remote target). The push driver's idle watchdog counts any
// frame from the sink as activity, so this stops a legitimately slow scan
// from being declared a stalled sink during a check-only run, where there
// is no DATA/ACK traffic to feed the watchdog. A sink that has actually
// died stops heartbeating and its pipe closes, so the watchdog still fires.
scanDone := make(chan struct{})
go func() {
t := time.NewTicker(10 * time.Second)
defer t.Stop()
for {
select {
case <-scanDone:
return
case <-t.C:
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "scan"})
}
}
}()
hashErrCh := make(chan error, 1)
go func() {
err := streamHashBlocks(f, size, blockSize, align, func(bh blockHash) error {
err := streamHashBlocks(f, base, size, blockSize, align, func(bh blockHash) error {
return out.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out))
close(scanDone)
if err == nil {
err = out.WriteFrame(frameHashDone, nil)
} else {
@ -533,7 +596,19 @@ func runSinkRole(path string, size, blockSize int64) error {
hashErrCh <- err
}()
loopErr := runDestLoop(destLoopParams{File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out})
var written int64
var lastWritten time.Time
loopErr := runDestLoop(destLoopParams{
File: f, BlockSize: blockSize, Align: align, In: in, AckOut: out,
OnWrite: func(uint64) {
written++
if time.Since(lastWritten) >= 500*time.Millisecond {
lastWritten = time.Now()
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
}
},
})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "write", Written: written})
if hashErr := <-hashErrCh; hashErr != nil {
return hashErr
}
@ -547,7 +622,7 @@ func runSinkRole(path string, size, blockSize int64) error {
// writing straight to its own stdout.
// ---------------------------------------------------------------------
func runSourceStreamRole(path string, size, blockSize int64) error {
func runSourceStreamRole(path string, base, size, blockSize int64) error {
out := NewFrameWriter(os.Stdout)
in := NewFrameReader(os.Stdin)
@ -599,21 +674,23 @@ func runSourceStreamRole(path string, size, blockSize int64) error {
}
}()
var copied, skipped int64
var copied, skipped, srcRead int64
var lastProgress time.Time
maybeProgress := func() {
if time.Since(lastProgress) < 500*time.Millisecond {
return
}
lastProgress = time.Now()
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: copied, Skipped: skipped, TotalBlocks: blockCount})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
}
loopErr := runSourceLoop(sourceLoopParams{
File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
OnRead: func(uint64) { srcRead++; maybeProgress() },
OnSkip: func(uint64) { skipped++; maybeProgress() },
OnSend: func(uint64, [32]byte) { copied++; maybeProgress() },
})
_ = out.WriteJSON(CtrlMsg{Type: msgProgress, Phase: "xfer", Copied: copied, Skipped: skipped, SrcRead: srcRead, TotalBlocks: blockCount})
if readErr := <-readErrCh; readErr != nil {
return readErr
}

View File

@ -4,18 +4,17 @@ set -e
cd "$(dirname "$0")"
mkdir -p dist
build_time=$(date -u +%Y-%m-%dT%H:%M:%SZ)
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" .
CGO_ENABLED=0 GOOS="$os" GOARCH="$arch" go build -trimpath \
-ldflags "-s -w -X main.buildTime=$build_time" -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/

View File

@ -19,8 +19,12 @@ type CtrlMsg struct {
// connect_push (-> source agent) / connect_pull (-> dest agent)
PeerHost string `json:"peerHost,omitempty"`
PeerUser string `json:"peerUser,omitempty"`
PeerPort int `json:"peerPort,omitempty"`
PeerPath string `json:"peerPath,omitempty"`
// Base/PeerBase shift every block offset on the local/peer side. Zero on
// both sides (the whole-file case) is what `sync` uses; the sink and
// source-stream roles still accept a non-zero base for a windowed copy.
Base int64 `json:"base,omitempty"`
PeerBase int64 `json:"peerBase,omitempty"`
// PeerLocal is set when both source and dest are local to the manager,
// so the driver (itself already a local child of the manager) can spawn
// the sink/source-stream helper as a plain local subprocess instead of
@ -39,18 +43,24 @@ type CtrlMsg struct {
Reason string `json:"reason,omitempty"`
Message string `json:"message,omitempty"`
// 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
// progress. Scan, source read and destination write all overlap and are
// tracked independently, so a run emits several kinds of "progress"
// message distinguished by Phase, interleaved; the printer keeps the
// latest of each:
// "scan" - Hashed/TotalBlocks: destination read throughput while it
// fingerprints its current content.
// "xfer" - Copied/Skipped/SrcRead/TotalBlocks: source read
// throughput as it reads and compares each block (Skipped:
// matched the destination; Copied: differed and was sent).
// "write" - Written: destination write throughput as changed blocks
// are actually applied.
Phase string `json:"phase,omitempty"`
Hashed int64 `json:"hashed,omitempty"`
Copied int64 `json:"copied,omitempty"`
Skipped int64 `json:"skipped,omitempty"`
SrcRead int64 `json:"srcRead,omitempty"`
Written int64 `json:"written,omitempty"`
TotalBlocks int64 `json:"totalBlocks,omitempty"`
BytesCopied int64 `json:"bytesCopied,omitempty"`
// log
Level string `json:"level,omitempty"`
}
const (
@ -65,7 +75,6 @@ const (
msgPullOK = "pull_ok"
msgPullFailed = "pull_failed"
msgProgress = "progress"
msgLog = "log"
msgError = "error"
msgClose = "close"
msgBye = "bye"

View File

@ -15,14 +15,31 @@ import (
// no install, no root.
const deployedRemoteBin = ".clonetool/bin/clonetool"
// resolveRemoteBin returns the path to a runnable clonetool on spec's host.
// It checks the configured --remote-bin first, then a copy deployed by an
// earlier run, and finally streams this binary over and verifies it runs.
// resolveRemoteBin returns the path to a clonetool on spec's host that is
// both runnable and the same build as this binary (compared via the
// "build=" tag in `version` output, see versionString/remoteBuildTag — an
// architecture mismatch alone, expected when cross-compiled, doesn't count
// as a mismatch). It checks the configured --remote-bin first, then a copy
// deployed by an earlier run, and finally streams this binary over and
// verifies it runs. With --deploy=false it can't correct a stale or missing
// binary, so it uses --remote-bin anyway if merely out of date (only a
// non-runnable one is still an error).
func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) {
if _, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
localTag := remoteBuildTag(versionString())
if out, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil {
return "", fmt.Errorf("%s: cannot reach %s over ssh: %w", tag, spec.Host, err)
} else if code == 0 {
return cfg.RemoteBin, nil
if remoteBuildTag(out) == localTag {
return cfg.RemoteBin, nil
}
if !cfg.Deploy {
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); continuing anyway (--deploy=false)\n",
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
return cfg.RemoteBin, nil
}
fmt.Fprintf(os.Stderr, "%s: %s on %s reports %q, a different build than this one (%s); looking for an up-to-date copy ...\n",
tag, cfg.RemoteBin, spec.Host, firstLine(out), versionString())
}
if !cfg.Deploy {
@ -30,11 +47,11 @@ func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) {
"install clonetool there or pass --remote-bin", tag, cfg.RemoteBin, spec.Host)
}
if _, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 {
if out, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 && remoteBuildTag(out) == localTag {
return deployedRemoteBin, nil
}
fmt.Fprintf(os.Stderr, "%s: no runnable clonetool on %s; copying this binary there ...\n", tag, spec.Host)
fmt.Fprintf(os.Stderr, "%s: no up-to-date clonetool on %s; copying this binary there ...\n", tag, spec.Host)
if err := deployBinary(cfg, spec); err != nil {
return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err)
}

View File

@ -13,20 +13,16 @@ type PathInfo struct {
Size int64
}
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
}
// alignmentFor returns the offset/length alignment a path's handle requires
// for positioned reads and writes. Linux 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).
func canElevate() bool { return true }
func statPath(path string) (PathInfo, error) {
fi, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
@ -63,9 +59,6 @@ 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)

View File

@ -1,9 +0,0 @@
//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 and windows (got path %s)", path)
}

View File

@ -1,18 +0,0 @@
//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 }

View File

@ -1,99 +0,0 @@
//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)
}

View File

@ -4,7 +4,6 @@ import (
"flag"
"fmt"
"os"
"runtime"
)
func main() {
@ -20,7 +19,7 @@ func main() {
case "agent":
err = cmdAgent(os.Args[2:])
case "version", "--version":
fmt.Printf("clonetool %s/%s\n", runtime.GOOS, runtime.GOARCH)
fmt.Println(versionString())
return
case "-h", "--help", "help":
usage()

View File

@ -153,11 +153,6 @@ 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()
@ -234,28 +229,62 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool {
return line == "y" || line == "yes"
}
// 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
// rateTracker smooths a monotonically increasing block counter into a
// blocks/sec rate, sampling no more than a few times a second so a burst of
// same-tick progress messages doesn't produce a noisy instantaneous rate.
type rateTracker struct {
lastT time.Time
lastVal int64
rate float64
}
func (r *rateTracker) update(val int64) {
now := time.Now()
if r.lastT.IsZero() {
r.lastT, r.lastVal = now, val
return
}
dt := now.Sub(r.lastT).Seconds()
if dt < 0.4 {
return
}
inst := float64(val-r.lastVal) / dt
if r.rate == 0 {
r.rate = inst
} else {
r.rate = 0.6*r.rate + 0.4*inst
}
r.lastT, r.lastVal = now, val
}
func (r *rateTracker) mib(blockSize int64) float64 {
return r.rate * float64(blockSize) / (1024 * 1024)
}
// progressPrinter renders a single rewriting status line on stderr. Scan,
// source read and destination write all run concurrently, each tracked by
// its own phase-tagged progress message (see CtrlMsg.Phase), so it keeps the
// latest of each and shows them on one line with their own rolling
// throughput; the "scan H/M" segment and its read rate disappear once the
// destination fingerprint is complete.
type progressPrinter struct {
blockSize int64
start time.Time
lastT time.Time
lastCopied int64
rate float64 // copied blocks/sec, smoothed
active bool
blockSize int64
active bool
hashed int64
copied int64
skipped int64
srcRead int64
written int64
total int64
dstReadRate rateTracker // destination scan/hash throughput
srcReadRate rateTracker // source read/compare throughput
dstWriteRate rateTracker // destination write throughput
}
func newProgressPrinter(blockSize int64) *progressPrinter {
now := time.Now()
return &progressPrinter{blockSize: blockSize, start: now, lastT: now}
return &progressPrinter{blockSize: blockSize}
}
func (p *progressPrinter) print(m CtrlMsg) {
@ -263,22 +292,22 @@ func (p *progressPrinter) print(m CtrlMsg) {
if m.TotalBlocks > 0 {
p.total = m.TotalBlocks
}
if m.Phase == "scan" {
switch m.Phase {
case "scan":
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
if p.total > 0 && p.hashed >= p.total {
p.dstReadRate.rate = 0 // scan finished; stop showing a frozen rate
} else {
p.dstReadRate.update(p.hashed)
}
case "xfer":
p.copied, p.skipped, p.srcRead = m.Copied, m.Skipped, m.SrcRead
p.srcReadRate.update(p.srcRead)
case "write":
p.written = m.Written
p.dstWriteRate.update(p.written)
}
done := p.copied + p.skipped
@ -286,15 +315,15 @@ func (p *progressPrinter) print(m CtrlMsg) {
if total <= 0 {
total = done
}
mib := p.rate * float64(p.blockSize) / (1024 * 1024)
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 ",
fmt.Fprintf(os.Stderr, "\r syncing %s %s/%s copied=%s skipped=%s%s rd(src) %5.1f rd(dst) %5.1f wr(dst) %5.1f MiB/s ",
progressBar(done, total), formatCount(done), formatCount(total),
formatCount(p.copied), formatCount(p.skipped), scan, mib)
formatCount(p.copied), formatCount(p.skipped), scan,
p.srcReadRate.mib(p.blockSize), p.dstReadRate.mib(p.blockSize), p.dstWriteRate.mib(p.blockSize))
}
// finish ends the current status line with a newline so following output

View File

@ -81,12 +81,14 @@ type blockHash struct {
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 {
// streamHashBlocks reads f in blockSize-byte blocks over the window
// [base, base+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. base is 0 for a whole-file sync; a non-zero base
// fingerprints just a byte window within a larger handle.
func streamHashBlocks(f *os.File, base, size, blockSize, align int64, onHash func(bh blockHash) error, onProgress func(done, total int64)) error {
if err := checkBlockAlign(blockSize, align); err != nil {
return err
}
@ -96,7 +98,7 @@ func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh b
onProgress(0, blockCount)
}
for i := int64(0); i < blockCount; i++ {
n, err := readBlockAt(f, buf, i*blockSize, size, align)
n, err := readBlockAt(f, buf, base+i*blockSize, base+size, align)
if err != nil {
return fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err)
}
@ -130,12 +132,18 @@ func scanProgressEmitter(out *FrameWriter) func(done, total int64) {
// never read twice.
type sourceLoopParams struct {
File *os.File // already opened for reading by the caller
Size int64
Base int64 // byte offset the window starts at (0 for a whole-file sync)
Size int64 // window length in bytes
BlockSize int64
Align int64
Hashes <-chan blockHash
Out *FrameWriter
OnSkip func(index uint64)
// OnRead is called immediately after a block has been read and hashed
// for comparison — whether it turns out to match or differ — so
// callers can track true source read throughput independent of the
// send/ack timing tracked by OnSkip/OnSend below.
OnRead func(index uint64)
OnSkip func(index uint64)
// 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
@ -151,12 +159,15 @@ func runSourceLoop(p sourceLoopParams) error {
f := p.File
buf := make([]byte, p.BlockSize)
for bh := range p.Hashes {
offset := int64(bh.index) * p.BlockSize
n, err := readBlockAt(f, buf, offset, p.Size, p.Align)
offset := p.Base + int64(bh.index)*p.BlockSize
n, err := readBlockAt(f, buf, offset, p.Base+p.Size, p.Align)
if err != nil {
return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err)
}
hash := sha256.Sum256(buf[:n])
if p.OnRead != nil {
p.OnRead(bh.index)
}
if hash == bh.hash {
if p.OnSkip != nil {
p.OnSkip(bh.index)
@ -181,6 +192,7 @@ func runSourceLoop(p sourceLoopParams) error {
// same process, before OnWritten is called).
type destLoopParams struct {
File *os.File // already opened for read/write by the caller
Base int64 // byte offset the window starts at (0 for a whole-file sync)
BlockSize int64
Align int64 // sector alignment for a raw disk handle, else 0/1
In *FrameReader
@ -189,6 +201,10 @@ type destLoopParams struct {
// pull mode, where the remote source-stream has no other channel back
// to the manager for progress updates). Sink-role callers leave it nil.
OnCtrlMsg func(CtrlMsg)
// OnWrite is called right after a block is durably written — the true
// destination write-throughput measurement point, as opposed to
// whatever "copied" count a driver derives from the send/ack protocol.
OnWrite func(index uint64)
}
func runDestLoop(p destLoopParams) error {
@ -212,12 +228,15 @@ func runDestLoop(p destLoopParams) error {
}
return fmt.Errorf("block %d: hash mismatch after transfer", index)
}
if err := writeBlockAt(f, block, int64(index)*p.BlockSize, p.Align); err != nil {
if err := writeBlockAt(f, block, p.Base+int64(index)*p.BlockSize, p.Align); err != nil {
if p.AckOut != nil {
_ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, err.Error()))
}
return fmt.Errorf("write block %d: %w", index, err)
}
if p.OnWrite != nil {
p.OnWrite(index)
}
if p.AckOut != nil {
if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil {
return err

16
util.go
View File

@ -2,9 +2,25 @@ package main
import (
"fmt"
"os/exec"
"strings"
"sync"
)
// runCmd runs name with args, returning trimmed combined output.
func runCmd(name string, args ...string) (string, error) {
out, err := exec.Command(name, args...).CombinedOutput()
return strings.TrimSpace(string(out)), err
}
// runCmdStdin is runCmd with a fixed stdin string (for tools that prompt).
func runCmdStdin(stdin, name string, args ...string) (string, error) {
c := exec.Command(name, args...)
c.Stdin = strings.NewReader(stdin)
out, err := c.CombinedOutput()
return strings.TrimSpace(string(out)), err
}
// limitedBuffer keeps only the last maxLen bytes written to it — used to
// capture a bounded tail of a subprocess's stderr for error messages
// without risking unbounded memory growth from a noisy remote command.

35
version.go Normal file
View File

@ -0,0 +1,35 @@
package main
import (
"fmt"
"runtime"
"strings"
)
// buildTime is stamped at build time via -ldflags "-X main.buildTime=..."
// (build.sh does this). A plain "go build" with no ldflags leaves it at
// "dev" — still a meaningful, comparable value: two "dev" builds are only
// ever the same binary bytes copied around (self-deploy), never two
// independently built ones landing on the same value by chance.
var buildTime = "dev"
// versionString is what `clonetool version` prints, and what a remote
// host's own `version` output is compared against (see remoteBuildTag) to
// tell whether it's running the same build as this binary.
func versionString() string {
return fmt.Sprintf("clonetool %s/%s build=%s", runtime.GOOS, runtime.GOARCH, buildTime)
}
// remoteBuildTag extracts the "build=..." tag from a clonetool `version`
// line, ignoring its GOOS/GOARCH — those are expected to legitimately
// differ across a cross-compiled deploy — and any surrounding ssh/shell
// noise on other lines. Returns "" if the line isn't a clonetool version
// line at all (e.g. "command not found").
func remoteBuildTag(out string) string {
line := firstLine(out)
i := strings.Index(line, "build=")
if i < 0 {
return ""
}
return strings.TrimSpace(line[i+len("build="):])
}