Compare commits

...

8 Commits
v0.1 ... main

14 changed files with 359 additions and 281 deletions

115
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 # clonetool
Block-level sync for a large file or block device between two machines (or 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" two paths on the same machine), driven from a third, passive "manager"
machine. Single static Go binary, no runtime dependencies beyond the 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 ## 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 is used at all — the source agent spawns the write-side helper as a
plain local subprocess. plain local subprocess.
- **Self-deploy:** if a remote endpoint has no runnable `clonetool` on - **Self-deploy:** if a remote endpoint has no runnable `clonetool` on
`PATH` (or wherever `--remote-bin` points), the manager streams *this* `PATH` (or wherever `--remote-bin` points), **or the one that's there is
binary to `~/.clonetool/bin/clonetool` on that host over the existing a different build than this binary** (see below), the manager streams
SSH connection and uses it — no install, no root. Disable with *this* binary to `~/.clonetool/bin/clonetool` on that host over the
`--deploy=false`. If the copied binary won't execute there (wrong CPU existing SSH connection and uses it — no install, no root. Disable with
architecture) the error says so; build one for the remote's arch `--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` (`CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build`) and put it on `PATH`
or pass `--remote-bin`. 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** - **sudo:** if reading or writing an endpoint that is a **block device**
fails with a permission error, `--sudo=auto` (the default) transparently fails with a permission error, `--sudo=auto` (the default) transparently
restarts that side's agent — and the helper it spawns on the peer — 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. start; `--sudo=never` never does.
- Sizing rules: - Sizing rules:
- Destination is a **block device**: it can't be resized, so if the - Destination is a **block device**: it can't be resized, so if the
source is larger the job fails; otherwise exactly `min(source, dest)` source is larger the job fails; otherwise exactly the source's size is
bytes are synced and the remainder of the device is left untouched. synced and the remainder of the device is left untouched.
- Destination is a **regular file**: it's truncated (created if - Destination is a **regular file**: it's truncated (created if
missing) to exactly the source's size, growing or shrinking it. missing) to exactly the source's size, growing or shrinking it.
Shrinking an existing non-empty file prompts for confirmation unless 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 . CGO_ENABLED=0 go build -o clonetool .
``` ```
Cross-compile for another OS/arch by setting `GOOS`/`GOARCH` (no cgo, so Cross-compile for another architecture by setting `GOARCH` (no cgo, so
these all work from any host): this works from any host):
``` ```
CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o clonetool . 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/`. `./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 Copy the resulting binary to the manager host. Source and destination
hosts get it automatically (see self-deploy above), or place it yourself hosts get it automatically (see self-deploy above), or place it yourself
and point `--remote-bin` at it. The binary is architecture-specific — and point `--remote-bin` at it. The binary is architecture-specific —
cross-compile if your hosts differ. Self-deploy streams a POSIX shell cross-compile if your hosts differ.
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 ## Usage
@ -136,6 +139,9 @@ clonetool sync --source LOC --dest LOC [options]
# Same machine # Same machine
clonetool sync --source /dev/sda --dest /dev/sdb 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 # Two remote machines, orchestrated from a third
clonetool sync --source db1:/dev/vdb --dest backup-host:/srv/db1.img 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. | | `--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). | | `--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 `clonetool version` prints the binary's `GOOS/GOARCH` and build timestamp,
the self-deploy check). 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 ## Caveats
- Block-device size detection is implemented on Linux (`BLKGETSIZE64`) and - Block-device size detection uses `BLKGETSIZE64`; the tool is Linux only.
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 - 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 file — clonetool won't create device nodes, so double-check device
paths for typos before running. 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)") role := fs.String("role", "", "control|sink|source-stream (internal)")
path := fs.String("path", "", "path to read/write") path := fs.String("path", "", "path to read/write")
size := fs.Int64("size", 0, "total sync size in bytes") 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") blockSize := fs.Int64("block-size", defaultBlockSize, "block size in bytes")
if err := fs.Parse(args); err != nil { if err := fs.Parse(args); err != nil {
return err return err
@ -28,11 +29,11 @@ func cmdAgent(args []string) error {
case "control": case "control":
return runControlAgent() return runControlAgent()
case "sink": case "sink":
return runSinkRole(*path, *size, *blockSize) return runSinkRole(*path, *base, *size, *blockSize)
case "source-stream": case "source-stream":
return runSourceStreamRole(*path, *size, *blockSize) return runSourceStreamRole(*path, *base, *size, *blockSize)
default: 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{ tailArgs := []string{
"agent", "--role", "sink", "agent", "--role", "sink",
"--path", req.PeerPath, "--path", req.PeerPath,
"--base", strconv.FormatInt(req.PeerBase, 10),
"--size", strconv.FormatInt(req.Size, 10), "--size", strconv.FormatInt(req.Size, 10),
"--block-size", strconv.FormatInt(req.BlockSize, 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 var pendingMu sync.Mutex
pending := make(map[uint64]bool) pending := make(map[uint64]bool)
var copied, skipped int64 var copied, skipped, srcRead int64
var lastProgress time.Time var lastProgress time.Time
maybeProgress := func() { maybeProgress := func() {
if time.Since(lastProgress) < 500*time.Millisecond { 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() lastProgress = time.Now()
_ = out.WriteJSON(CtrlMsg{ _ = 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, 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 // hashCh is sized to hold every block hash so the reader below never
// blocks handing hashes off (which, since ACKs share the same stream, // blocks handing hashes off (which, since ACKs share the same stream,
// would otherwise be able to deadlock against the in-flight-send limit). // 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 return
} }
markActivity()
switch typ { switch typ {
case frameBlockHash: case frameBlockHash:
idx, h, derr := decodeBlockHashFrame(payload) idx, h, derr := decodeBlockHashFrame(payload)
@ -315,8 +331,10 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
go func() { go func() {
sendErrCh <- runSourceLoop(sourceLoopParams{ sendErrCh <- runSourceLoop(sourceLoopParams{
File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Align: align, Hashes: hashCh, Out: fw, 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) { OnSend: func(idx uint64, _ [32]byte) {
markActivity()
sem <- struct{}{} sem <- struct{}{}
pendingMu.Lock() pendingMu.Lock()
pending[idx] = true pending[idx] = true
@ -332,7 +350,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
sendCh := sendErrCh sendCh := sendErrCh
ackCh := ackEvents ackCh := ackEvents
var fatalErr error 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() defer idle.Stop()
for sendCh != nil || ackCh != nil { for sendCh != nil || ackCh != nil {
if !idle.Stop() { if !idle.Stop() {
@ -341,7 +363,7 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
default: default:
} }
} }
idle.Reset(120 * time.Second) idle.Reset(idleTimeout)
select { select {
case sendErr := <-sendCh: case sendErr := <-sendCh:
sendCh = nil sendCh = nil
@ -367,7 +389,11 @@ func pumpPush(req CtrlMsg, align int64, srcFile *os.File, fw *FrameWriter, fr *F
maybeProgress() maybeProgress()
} }
case <-idle.C: 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 { if fatalErr != nil {
break 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) 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 return nil
} }
@ -400,6 +430,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
tailArgs := []string{ tailArgs := []string{
"agent", "--role", "source-stream", "agent", "--role", "source-stream",
"--path", req.PeerPath, "--path", req.PeerPath,
"--base", strconv.FormatInt(req.PeerBase, 10),
"--size", strconv.FormatInt(req.Size, 10), "--size", strconv.FormatInt(req.Size, 10),
"--block-size", strconv.FormatInt(req.BlockSize, 10), "--block-size", strconv.FormatInt(req.BlockSize, 10),
} }
@ -448,7 +479,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
hashErrCh := make(chan error, 1) hashErrCh := make(chan error, 1)
go func() { 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)) return fw.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out)) }, scanProgressEmitter(out))
if err == nil { if err == nil {
@ -462,14 +493,24 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
hashErrCh <- err hashErrCh <- err
}() }()
var written int64
var lastWritten time.Time
loopErr := runDestLoop(destLoopParams{ 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) { OnCtrlMsg: func(m CtrlMsg) {
if m.Type == msgProgress { if m.Type == msgProgress {
_ = out.WriteJSON(m) _ = 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 hashErr := <-hashErrCh
if hashErr != nil { if hashErr != nil {
@ -496,7 +537,7 @@ func runPullDriver(req CtrlMsg, out *FrameWriter) {
// destination host. Dumb write endpoint: verify+pwrite+ack per block. // 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) out := NewFrameWriter(os.Stdout)
in := NewFrameReader(os.Stdin) 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 // 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 // only sends after seeing that block's hash), so the two accesses to f
// never race on the same region. // 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) hashErrCh := make(chan error, 1)
go func() { 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)) return out.WriteFrame(frameBlockHash, encodeBlockHashFrame(bh.index, bh.hash))
}, scanProgressEmitter(out)) }, scanProgressEmitter(out))
close(scanDone)
if err == nil { if err == nil {
err = out.WriteFrame(frameHashDone, nil) err = out.WriteFrame(frameHashDone, nil)
} else { } else {
@ -533,7 +596,19 @@ func runSinkRole(path string, size, blockSize int64) error {
hashErrCh <- err 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 { if hashErr := <-hashErrCh; hashErr != nil {
return hashErr return hashErr
} }
@ -547,7 +622,7 @@ func runSinkRole(path string, size, blockSize int64) error {
// writing straight to its own stdout. // 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) out := NewFrameWriter(os.Stdout)
in := NewFrameReader(os.Stdin) 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 var lastProgress time.Time
maybeProgress := func() { maybeProgress := func() {
if time.Since(lastProgress) < 500*time.Millisecond { if time.Since(lastProgress) < 500*time.Millisecond {
return return
} }
lastProgress = time.Now() 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{ loopErr := runSourceLoop(sourceLoopParams{
File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out, File: f, Size: size, BlockSize: blockSize, Align: align, Hashes: hashCh, Out: out,
OnRead: func(uint64) { srcRead++; maybeProgress() },
OnSkip: func(uint64) { skipped++; maybeProgress() }, OnSkip: func(uint64) { skipped++; maybeProgress() },
OnSend: func(uint64, [32]byte) { copied++; 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 { if readErr := <-readErrCh; readErr != nil {
return readErr return readErr
} }

View File

@ -4,18 +4,17 @@ set -e
cd "$(dirname "$0")" cd "$(dirname "$0")"
mkdir -p dist mkdir -p dist
build_time=$(date -u +%Y-%m-%dT%H:%M:%SZ)
build() { build() {
os=$1 arch=$2 out=$3 os=$1 arch=$2 out=$3
echo "==> $os/$arch -> dist/$out" 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 amd64 clonetool-linux-amd64
build linux arm64 clonetool-linux-arm64 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:" echo "done:"
ls -la dist/ ls -la dist/

View File

@ -19,8 +19,12 @@ type CtrlMsg struct {
// connect_push (-> source agent) / connect_pull (-> dest agent) // connect_push (-> source agent) / connect_pull (-> dest agent)
PeerHost string `json:"peerHost,omitempty"` PeerHost string `json:"peerHost,omitempty"`
PeerUser string `json:"peerUser,omitempty"` PeerUser string `json:"peerUser,omitempty"`
PeerPort int `json:"peerPort,omitempty"`
PeerPath string `json:"peerPath,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, // 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 // 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 // the sink/source-stream helper as a plain local subprocess instead of
@ -39,18 +43,24 @@ type CtrlMsg struct {
Reason string `json:"reason,omitempty"` Reason string `json:"reason,omitempty"`
Message string `json:"message,omitempty"` Message string `json:"message,omitempty"`
// progress. Scan and transfer now overlap, so a run emits both "scan" // progress. Scan, source read and destination write all overlap and are
// messages (Hashed/TotalBlocks) and transfer messages (Copied/Skipped) // tracked independently, so a run emits several kinds of "progress"
// interleaved; the printer keeps the latest of each. // message distinguished by Phase, interleaved; the printer keeps the
Phase string `json:"phase,omitempty"` // "scan" while hashing the destination; empty during transfer // 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"` Hashed int64 `json:"hashed,omitempty"`
Copied int64 `json:"copied,omitempty"` Copied int64 `json:"copied,omitempty"`
Skipped int64 `json:"skipped,omitempty"` Skipped int64 `json:"skipped,omitempty"`
SrcRead int64 `json:"srcRead,omitempty"`
Written int64 `json:"written,omitempty"`
TotalBlocks int64 `json:"totalBlocks,omitempty"` TotalBlocks int64 `json:"totalBlocks,omitempty"`
BytesCopied int64 `json:"bytesCopied,omitempty"`
// log
Level string `json:"level,omitempty"`
} }
const ( const (
@ -65,7 +75,6 @@ const (
msgPullOK = "pull_ok" msgPullOK = "pull_ok"
msgPullFailed = "pull_failed" msgPullFailed = "pull_failed"
msgProgress = "progress" msgProgress = "progress"
msgLog = "log"
msgError = "error" msgError = "error"
msgClose = "close" msgClose = "close"
msgBye = "bye" msgBye = "bye"

View File

@ -15,26 +15,43 @@ import (
// no install, no root. // no install, no root.
const deployedRemoteBin = ".clonetool/bin/clonetool" const deployedRemoteBin = ".clonetool/bin/clonetool"
// resolveRemoteBin returns the path to a runnable clonetool on spec's host. // resolveRemoteBin returns the path to a clonetool on spec's host that is
// It checks the configured --remote-bin first, then a copy deployed by an // both runnable and the same build as this binary (compared via the
// earlier run, and finally streams this binary over and verifies it runs. // "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) { 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) return "", fmt.Errorf("%s: cannot reach %s over ssh: %w", tag, spec.Host, err)
} else if code == 0 { } else if code == 0 {
if remoteBuildTag(out) == localTag {
return cfg.RemoteBin, nil 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 { if !cfg.Deploy {
return "", fmt.Errorf("%s: %q is not runnable on %s and --deploy=false; "+ return "", fmt.Errorf("%s: %q is not runnable on %s and --deploy=false; "+
"install clonetool there or pass --remote-bin", tag, cfg.RemoteBin, spec.Host) "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 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 { if err := deployBinary(cfg, spec); err != nil {
return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err) return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err)
} }

View File

@ -13,20 +13,16 @@ type PathInfo struct {
Size int64 Size int64
} }
func statPath(path string) (PathInfo, error) { // alignmentFor returns the offset/length alignment a path's handle requires
// A raw disk handle (Windows \\.\PhysicalDrive0, \\.\C:) is not something // for positioned reads and writes. Linux block devices accept ordinary
// os.Stat can describe, so ask the platform for its size directly. // buffered pread/pwrite at any alignment, so there is nothing to round to.
if isDevicePath(path) { func alignmentFor(string) int64 { return 1 }
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
}
// 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) fi, err := os.Stat(path)
if err != nil { if err != nil {
if os.IsNotExist(err) { if os.IsNotExist(err) {
@ -63,9 +59,6 @@ func prepareDest(path string, targetSize int64) error {
} }
return nil 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) f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644)
if err != nil { if err != nil {
return fmt.Errorf("open/create %s: %w", path, err) 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" "flag"
"fmt" "fmt"
"os" "os"
"runtime"
) )
func main() { func main() {
@ -20,7 +19,7 @@ func main() {
case "agent": case "agent":
err = cmdAgent(os.Args[2:]) err = cmdAgent(os.Args[2:])
case "version", "--version": case "version", "--version":
fmt.Printf("clonetool %s/%s\n", runtime.GOOS, runtime.GOARCH) fmt.Println(versionString())
return return
case "-h", "--help", "help": case "-h", "--help", "help":
usage() usage()

View File

@ -153,11 +153,6 @@ func bringUpController(spec Spec, tag string, cfg *SyncConfig, remoteBin, probeP
if err == nil { if err == nil {
return c, info, sudo, 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 { if errors.Is(err, errNeedPriv) && cfg.Sudo == "auto" && !sudo {
fmt.Fprintf(os.Stderr, "%s: permission denied on %s; retrying via sudo ...\n", tag, probePath) fmt.Fprintf(os.Stderr, "%s: permission denied on %s; retrying via sudo ...\n", tag, probePath)
c.Close() c.Close()
@ -234,28 +229,62 @@ func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool {
return line == "y" || line == "yes" return line == "y" || line == "yes"
} }
// progressPrinter renders a single rewriting status line on stderr. Scan // rateTracker smooths a monotonically increasing block counter into a
// and transfer now run concurrently, so it keeps the latest of each kind of // blocks/sec rate, sampling no more than a few times a second so a burst of
// update (blocks hashed so far, blocks copied/skipped) and shows them on one // same-tick progress messages doesn't produce a noisy instantaneous rate.
// line, with a rolling copy rate; the "scan H/M" segment disappears once the 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. // destination fingerprint is complete.
type progressPrinter struct { type progressPrinter struct {
blockSize int64 blockSize int64
start time.Time
lastT time.Time
lastCopied int64
rate float64 // copied blocks/sec, smoothed
active bool active bool
hashed int64 hashed int64
copied int64 copied int64
skipped int64 skipped int64
srcRead int64
written int64
total 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 { func newProgressPrinter(blockSize int64) *progressPrinter {
now := time.Now() return &progressPrinter{blockSize: blockSize}
return &progressPrinter{blockSize: blockSize, start: now, lastT: now}
} }
func (p *progressPrinter) print(m CtrlMsg) { func (p *progressPrinter) print(m CtrlMsg) {
@ -263,22 +292,22 @@ func (p *progressPrinter) print(m CtrlMsg) {
if m.TotalBlocks > 0 { if m.TotalBlocks > 0 {
p.total = m.TotalBlocks p.total = m.TotalBlocks
} }
if m.Phase == "scan" { switch m.Phase {
case "scan":
if m.Hashed > p.hashed { if m.Hashed > p.hashed {
p.hashed = m.Hashed p.hashed = m.Hashed
} }
if p.total > 0 && p.hashed >= p.total {
p.dstReadRate.rate = 0 // scan finished; stop showing a frozen rate
} else { } else {
p.copied, p.skipped = m.Copied, m.Skipped p.dstReadRate.update(p.hashed)
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
} }
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 done := p.copied + p.skipped
@ -286,15 +315,15 @@ func (p *progressPrinter) print(m CtrlMsg) {
if total <= 0 { if total <= 0 {
total = done total = done
} }
mib := p.rate * float64(p.blockSize) / (1024 * 1024)
scan := "" scan := ""
if p.total > 0 && p.hashed < p.total { if p.total > 0 && p.hashed < p.total {
scan = fmt.Sprintf(" scan %s/%s", formatCount(p.hashed), formatCount(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), 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 // finish ends the current status line with a newline so following output

View File

@ -81,12 +81,14 @@ type blockHash struct {
hash [32]byte hash [32]byte
} }
// streamHashBlocks reads f in blockSize-byte blocks up to size, hashing each // streamHashBlocks reads f in blockSize-byte blocks over the window
// one and handing it to onHash the moment it is computed (this is what lets // [base, base+size), hashing each one and handing it to onHash the moment it
// the peer start comparing before the whole side has been scanned). It keeps // is computed (this is what lets the peer start comparing before the whole
// no hash state of its own. onProgress, if non-nil, is called with // side has been scanned). It keeps no hash state of its own. onProgress, if
// (blocksHashed, totalBlocks) before the first block and after each one. // non-nil, is called with (blocksHashed, totalBlocks) before the first block
func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh blockHash) error, onProgress func(done, total int64)) error { // 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 { if err := checkBlockAlign(blockSize, align); err != nil {
return err return err
} }
@ -96,7 +98,7 @@ func streamHashBlocks(f *os.File, size, blockSize, align int64, onHash func(bh b
onProgress(0, blockCount) onProgress(0, blockCount)
} }
for i := int64(0); i < blockCount; i++ { 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 { if err != nil {
return 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)
} }
@ -130,11 +132,17 @@ func scanProgressEmitter(out *FrameWriter) func(done, total int64) {
// never read twice. // never read twice.
type sourceLoopParams struct { type sourceLoopParams struct {
File *os.File // already opened for reading by the caller 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 BlockSize int64
Align int64 Align int64
Hashes <-chan blockHash Hashes <-chan blockHash
Out *FrameWriter Out *FrameWriter
// 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) OnSkip func(index uint64)
// OnSend is called for a differing block immediately before its DATA // OnSend is called for a differing block immediately before its DATA
// frame goes on the wire — never after. The push driver relies on this // frame goes on the wire — never after. The push driver relies on this
@ -151,12 +159,15 @@ func runSourceLoop(p sourceLoopParams) error {
f := p.File f := p.File
buf := make([]byte, p.BlockSize) buf := make([]byte, p.BlockSize)
for bh := range p.Hashes { for bh := range p.Hashes {
offset := int64(bh.index) * p.BlockSize offset := p.Base + int64(bh.index)*p.BlockSize
n, err := readBlockAt(f, buf, offset, p.Size, p.Align) n, err := readBlockAt(f, buf, offset, p.Base+p.Size, p.Align)
if err != nil { if err != nil {
return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err) return fmt.Errorf("read %s at block %d: %w", f.Name(), bh.index, err)
} }
hash := sha256.Sum256(buf[:n]) hash := sha256.Sum256(buf[:n])
if p.OnRead != nil {
p.OnRead(bh.index)
}
if hash == bh.hash { if hash == bh.hash {
if p.OnSkip != nil { if p.OnSkip != nil {
p.OnSkip(bh.index) p.OnSkip(bh.index)
@ -181,6 +192,7 @@ func runSourceLoop(p sourceLoopParams) error {
// same process, before OnWritten is called). // same process, before OnWritten is called).
type destLoopParams struct { type destLoopParams struct {
File *os.File // already opened for read/write by the caller 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 BlockSize int64
Align int64 // sector alignment for a raw disk handle, else 0/1 Align int64 // sector alignment for a raw disk handle, else 0/1
In *FrameReader In *FrameReader
@ -189,6 +201,10 @@ type destLoopParams struct {
// pull mode, where the remote source-stream has no other channel back // pull mode, where the remote source-stream has no other channel back
// to the manager for progress updates). Sink-role callers leave it nil. // to the manager for progress updates). Sink-role callers leave it nil.
OnCtrlMsg func(CtrlMsg) 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 { func runDestLoop(p destLoopParams) error {
@ -212,12 +228,15 @@ func runDestLoop(p destLoopParams) error {
} }
return fmt.Errorf("block %d: hash mismatch after transfer", index) 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 { if p.AckOut != nil {
_ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, err.Error())) _ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, err.Error()))
} }
return fmt.Errorf("write block %d: %w", index, err) return fmt.Errorf("write block %d: %w", index, err)
} }
if p.OnWrite != nil {
p.OnWrite(index)
}
if p.AckOut != nil { if p.AckOut != nil {
if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil { if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil {
return err return err

16
util.go
View File

@ -2,9 +2,25 @@ package main
import ( import (
"fmt" "fmt"
"os/exec"
"strings"
"sync" "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 // limitedBuffer keeps only the last maxLen bytes written to it — used to
// capture a bounded tail of a subprocess's stderr for error messages // capture a bounded tail of a subprocess's stderr for error messages
// without risking unbounded memory growth from a noisy remote command. // 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="):])
}