commit 9e5e0fa8312d342c0954048949bc7005cee356d4 Author: Alexander Gabriel Date: Sat Sep 5 23:02:29 2026 +0200 first batch diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6490ecf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/clonetool diff --git a/README.md b/README.md new file mode 100644 index 0000000..da611fb --- /dev/null +++ b/README.md @@ -0,0 +1,103 @@ +# 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. + +## How it works + +- You run `clonetool sync` on the **manager**. Source, destination, and + manager may be three entirely different machines — the manager never + reads or writes a single block itself. It only spawns and talks to two + **control agents** (`clonetool agent --role control`, one per side, + local subprocess or over `ssh`) that do the stat/prepare/transfer work. +- The two agents compare content by **SHA-256 hash per fixed-size block**. + A block is only re-read once: it's hashed from the buffer it was read + into, and that same buffer is what gets sent on if it differs — never + read twice. +- Bulk data goes **directly between source and destination**, not through + the manager. Each run tries: + 1. **push** — the source agent connects straight to the destination + host over SSH and streams changed blocks to it. + 2. **pull** (fallback) — if push isn't possible (no route/keys in that + direction), the destination agent connects to the source host + instead and pulls. + 3. If neither direction works, the job fails with both reasons named. + Run the manager on the source or destination host, or set up SSH + connectivity in at least one direction. + - When source and destination are **both local** to the manager, no SSH + is used at all — the source agent spawns the write-side helper as a + plain local subprocess. +- Each named **job** keeps a hash table for the destination + (`~/.clonetool/jobs/.state` by default) so a re-run doesn't need to + read the destination again — it just compares freshly-hashed source + blocks against last known state. **The destination must not be modified + by anything else between syncs of the same job** — that assumption is + what lets the tool skip reading it. Running the same job against a + different destination is refused (pass `--force` to deliberately rebind + it, which discards the hash history). +- 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. + - 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 + `--yes` is passed. + +## Build + +``` +CGO_ENABLED=0 go build -o clonetool . +``` + +Copy the resulting binary to the manager, source, and destination hosts +(same path, or point `--remote-bin` at wherever it lives on each host — +clonetool does not deploy itself). + +## Usage + +``` +clonetool sync --job NAME --source LOC --dest LOC [options] +``` + +`LOC` is either a local path (`/dev/sdb`, `./image.bin`) or +`[user@]host:path` for a path reached over SSH. + +``` +# Same machine +clonetool sync --job disk1 --source /dev/sda --dest /dev/sdb + +# Two remote machines, orchestrated from a third +clonetool sync --job backup --source db1:/dev/vdb --dest backup-host:/srv/db1.img + +# Re-run any time; only changed blocks move +clonetool sync --job backup --source db1:/dev/vdb --dest backup-host:/srv/db1.img +``` + +Options: + +| Flag | Default | Meaning | +|---|---|---| +| `--block-size` | `4M` | Block size (accepts `K`/`M`/`G` suffixes). Changing it on an existing job discards its hash history. | +| `--state-dir` | `~/.clonetool/jobs` | Where job hash-tables live. | +| `--yes` | off | Don't prompt before shrinking an existing destination file. | +| `--force` | off | Rebind a job to a different source/dest, discarding its hash history. | +| `--connect-timeout` | `8` | SSH connect timeout (seconds) used for the push/pull direction probe. | +| `--ssh` | `ssh` | ssh binary to use. | +| `--ssh-opt` | — | Extra `-o OPT` passed to ssh (repeatable). | +| `--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). | + +## Caveats + +- Block-device size detection (`BLKGETSIZE64`) 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. +- SSH host keys are accepted on first connect (`StrictHostKeyChecking=accept-new`) + and rejected if they later change, same as normal SSH behavior. +- `agent` is an internal subcommand spawned automatically by `sync`; it's + not meant to be run by hand, though it will work standalone for + debugging. diff --git a/agent.go b/agent.go new file mode 100644 index 0000000..ce797d2 --- /dev/null +++ b/agent.go @@ -0,0 +1,557 @@ +package main + +import ( + "encoding/hex" + "encoding/json" + "flag" + "fmt" + "io" + "os" + "os/exec" + "strconv" + "sync" + "sync/atomic" + "time" +) + +func cmdAgent(args []string) error { + fs := flag.NewFlagSet("agent", flag.ContinueOnError) + 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") + blockSize := fs.Int64("block-size", defaultBlockSize, "block size in bytes") + if err := fs.Parse(args); err != nil { + return err + } + + switch *role { + case "control": + return runControlAgent() + case "sink": + return runSinkRole(*path, *size, *blockSize) + case "source-stream": + return runSourceStreamRole(*path, *size, *blockSize) + default: + return fmt.Errorf("agent: unknown or missing --role %q (want control|sink|source-stream)", *role) + } +} + +// --------------------------------------------------------------------- +// control role: long-lived per-side orchestration agent, driven by the +// manager over stdin/stdout with CtrlMsg frames. +// --------------------------------------------------------------------- + +func runControlAgent() error { + in := NewFrameReader(os.Stdin) + out := NewFrameWriter(os.Stdout) + + for { + typ, payload, err := in.ReadFrame() + if err != nil { + return nil // manager closed the pipe; nothing left to do + } + if typ != frameCtrlJSON { + return fmt.Errorf("control agent: unexpected frame type %d", typ) + } + var m CtrlMsg + if err := json.Unmarshal(payload, &m); err != nil { + return fmt.Errorf("control agent: decode message: %w", err) + } + switch m.Type { + case msgStat: + handleStat(out, m) + case msgPrepare: + handlePrepare(out, m) + case msgConnectPush: + hashes, err := readHashTable(in) + if err != nil { + return err + } + runPushDriver(m, hashes, out) + case msgConnectPull: + hashes, err := readHashTable(in) + if err != nil { + return err + } + runPullDriver(m, hashes, out) + case msgClose: + _ = out.WriteJSON(CtrlMsg{Type: msgBye}) + return nil + default: + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("unknown command %q", m.Type)}) + } + } +} + +func readHashTable(in *FrameReader) ([][32]byte, error) { + typ, payload, err := in.ReadFrame() + if err != nil { + return nil, fmt.Errorf("read hash table: %w", err) + } + if typ != frameHashTable { + return nil, fmt.Errorf("expected hash table frame, got type %d", typ) + } + return unflattenHashes(payload) +} + +func handleStat(out *FrameWriter, m CtrlMsg) { + info, err := statPath(m.Path) + if err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + return + } + _ = out.WriteJSON(CtrlMsg{Type: msgStatOK, Exists: info.Exists, IsDevice: info.IsDevice, Size: info.Size}) +} + +func handlePrepare(out *FrameWriter, m CtrlMsg) { + if err := prepareDest(m.Path, m.Size); err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + return + } + _ = out.WriteJSON(CtrlMsg{Type: msgPrepareOK, Size: m.Size}) +} + +// --------------------------------------------------------------------- +// push driver: runs inside the SOURCE control agent. Spawns ssh straight +// to the destination host, running the "sink" role, and — once it answers +// READY — performs the whole read/hash/compare/send loop itself. +// --------------------------------------------------------------------- + +func runPushDriver(req CtrlMsg, hashes [][32]byte, out *FrameWriter) { + tailArgs := []string{ + "agent", "--role", "sink", + "--path", req.PeerPath, + "--size", strconv.FormatInt(req.Size, 10), + "--block-size", strconv.FormatInt(req.BlockSize, 10), + } + var cmd *exec.Cmd + if req.PeerLocal { + cmd = localAgentCommand(tailArgs) + } else { + cmd = sshCommand(req.SSHBin, req.SSHOpts, true, req.ConnectTimeoutSec, req.PeerUser, req.PeerHost, append([]string{req.RemoteBin}, tailArgs...)) + } + stdin, err := cmd.StdinPipe() + if err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPushFailed, Reason: err.Error()}) + return + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPushFailed, Reason: err.Error()}) + return + } + stderrBuf := newLimitedBuffer(4096) + cmd.Stderr = stderrBuf + + if err := cmd.Start(); err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPushFailed, Reason: fmt.Sprintf("start ssh: %v", err)}) + return + } + fw := NewFrameWriter(stdin) + fr := NewFrameReader(stdout) + + timeout := time.Duration(req.ConnectTimeoutSec+2) * time.Second + if err := waitReady(fr, timeout); err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgPushFailed, Reason: fmt.Sprintf("%v (remote stderr: %s)", err, stderrBuf.String())}) + return + } + + // Handshake succeeded: we're committed to push for this run. + srcFile, err := os.Open(req.Path) + if err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + return + } + defer srcFile.Close() + + if fatalErr := pumpPush(req, hashes, srcFile, fw, fr, out); fatalErr != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fatalErr.Error()}) + return + } + if err := cmd.Wait(); err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("sink process: %v (stderr: %s)", err, stderrBuf.String())}) + return + } + _ = out.WriteJSON(CtrlMsg{Type: msgPushOK}) +} + +func waitReady(fr *FrameReader, timeout time.Duration) error { + type result struct { + typ frameType + err error + } + ch := make(chan result, 1) + go func() { + typ, _, err := fr.ReadFrame() + ch <- result{typ, err} + }() + select { + case r := <-ch: + if r.err != nil { + return fmt.Errorf("handshake failed: %w", r.err) + } + if r.typ != frameReady { + return fmt.Errorf("handshake failed: unexpected frame type %d", r.typ) + } + return nil + case <-time.After(timeout): + return fmt.Errorf("handshake timed out after %s", timeout) + } +} + +type ackEvent struct { + index uint64 + eof bool + err error +} + +// pumpPush runs the source-side read/hash/compare/send loop against fw +// (the pipe to the remote sink) while concurrently draining ACK/ERR frames +// from fr, only reporting a block as done to the manager (over out) once +// its write has been confirmed. +func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, fr *FrameReader, out *FrameWriter) error { + const maxInFlight = 32 + sem := make(chan struct{}, maxInFlight) + + var pendingMu sync.Mutex + pending := make(map[uint64][32]byte) + + batch := newResultBatcher(out) + defer batch.flush() + + var copied, skipped int64 + var lastProgress time.Time + maybeProgress := func() { + if time.Since(lastProgress) < 500*time.Millisecond { + return + } + lastProgress = time.Now() + _ = out.WriteJSON(CtrlMsg{ + Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), + TotalBlocks: int64(len(hashes)), + }) + } + + ackEvents := make(chan ackEvent, 256) + go func() { + for { + typ, payload, err := fr.ReadFrame() + if err != nil { + if err == io.EOF { + ackEvents <- ackEvent{eof: true} + } else { + ackEvents <- ackEvent{err: err} + } + return + } + switch typ { + case frameAck: + idx, err := decodeIndexFrame(payload) + if err != nil { + ackEvents <- ackEvent{err: err} + return + } + ackEvents <- ackEvent{index: idx} + case frameErr: + idx, msg, _ := decodeErrFrame(payload) + ackEvents <- ackEvent{err: fmt.Errorf("remote reported error at block %d: %s", idx, msg)} + return + default: + ackEvents <- ackEvent{err: fmt.Errorf("unexpected frame type %d from sink", typ)} + return + } + } + }() + + sendErrCh := make(chan error, 1) + go func() { + sendErrCh <- runSourceLoop(sourceLoopParams{ + File: srcFile, Size: req.Size, BlockSize: req.BlockSize, Hashes: hashes, Out: fw, + OnSkip: func(uint64) { atomic.AddInt64(&skipped, 1); maybeProgress() }, + OnSend: func(idx uint64, hash [32]byte) { + sem <- struct{}{} + pendingMu.Lock() + pending[idx] = hash + pendingMu.Unlock() + }, + }) + }() + + // Wait for both: the send loop to finish (all DATA frames + DONE sent) + // and the ack stream to end. Sink closes its stdout (a clean EOF) only + // after it has acked every block it received, so an EOF while blocks + // are still unconfirmed is treated as a real failure below. + sendCh := sendErrCh + ackCh := ackEvents + var fatalErr error + idle := time.NewTimer(120 * time.Second) + defer idle.Stop() + for sendCh != nil || ackCh != nil { + if !idle.Stop() { + select { + case <-idle.C: + default: + } + } + idle.Reset(120 * time.Second) + select { + case sendErr := <-sendCh: + sendCh = nil + if sendErr != nil { + fatalErr = sendErr + } + case ev := <-ackCh: + switch { + case ev.err != nil: + fatalErr = ev.err + ackCh = nil + case ev.eof: + ackCh = nil + default: + pendingMu.Lock() + h, ok := pending[ev.index] + delete(pending, ev.index) + pendingMu.Unlock() + if ok { + atomic.AddInt64(&copied, 1) + batch.add(BlockResult{Index: ev.index, Hash: hex.EncodeToString(h[:])}) + } + <-sem + maybeProgress() + } + case <-idle.C: + fatalErr = fmt.Errorf("timed out waiting for the sink") + } + if fatalErr != nil { + break + } + } + if fatalErr != nil { + return fatalErr + } + + pendingMu.Lock() + n := len(pending) + pendingMu.Unlock() + if n > 0 { + return fmt.Errorf("sink closed the connection with %d block write confirmation(s) still outstanding", n) + } + + batch.flush() + _ = out.WriteJSON(CtrlMsg{Type: msgProgress, Copied: atomic.LoadInt64(&copied), Skipped: atomic.LoadInt64(&skipped), TotalBlocks: int64(len(hashes))}) + return nil +} + +// --------------------------------------------------------------------- +// pull driver: runs inside the DEST control agent. Spawns ssh straight to +// the source host, running the "source-stream" role, feeds it the hash +// table, then writes whatever it streams back directly to the local +// destination — no round trip needed to confirm a write, since dest-agent +// itself performed it. +// --------------------------------------------------------------------- + +func runPullDriver(req CtrlMsg, hashes [][32]byte, out *FrameWriter) { + tailArgs := []string{ + "agent", "--role", "source-stream", + "--path", req.PeerPath, + "--size", strconv.FormatInt(req.Size, 10), + "--block-size", strconv.FormatInt(req.BlockSize, 10), + } + var cmd *exec.Cmd + if req.PeerLocal { + cmd = localAgentCommand(tailArgs) + } else { + cmd = sshCommand(req.SSHBin, req.SSHOpts, true, req.ConnectTimeoutSec, req.PeerUser, req.PeerHost, append([]string{req.RemoteBin}, tailArgs...)) + } + stdin, err := cmd.StdinPipe() + if err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPullFailed, Reason: err.Error()}) + return + } + stdout, err := cmd.StdoutPipe() + if err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPullFailed, Reason: err.Error()}) + return + } + stderrBuf := newLimitedBuffer(4096) + cmd.Stderr = stderrBuf + + if err := cmd.Start(); err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgPullFailed, Reason: fmt.Sprintf("start ssh: %v", err)}) + return + } + fw := NewFrameWriter(stdin) + fr := NewFrameReader(stdout) + + timeout := time.Duration(req.ConnectTimeoutSec+2) * time.Second + if err := waitReady(fr, timeout); err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgPullFailed, Reason: fmt.Sprintf("%v (remote stderr: %s)", err, stderrBuf.String())}) + return + } + + // Handshake succeeded: send the hash table and commit to pull. + if err := fw.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("send hash table: %v", err)}) + return + } + + dstFile, err := os.OpenFile(req.Path, os.O_RDWR, 0) + if err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + return + } + defer dstFile.Close() + + batch := newResultBatcher(out) + var copied int64 + loopErr := runDestLoop(destLoopParams{ + File: dstFile, BlockSize: req.BlockSize, In: fr, + OnWritten: func(idx uint64, hash [32]byte) { + atomic.AddInt64(&copied, 1) + batch.add(BlockResult{Index: idx, Hash: hex.EncodeToString(hash[:])}) + }, + OnCtrlMsg: func(m CtrlMsg) { + if m.Type == msgProgress { + _ = out.WriteJSON(m) + } + }, + }) + batch.flush() + + if loopErr != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: loopErr.Error()}) + return + } + if err := cmd.Wait(); err != nil { + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("source-stream process: %v (stderr: %s)", err, stderrBuf.String())}) + return + } + _ = out.WriteJSON(CtrlMsg{Type: msgPullOK}) +} + +// --------------------------------------------------------------------- +// sink role: one-shot process spawned (over ssh, in push mode) on the +// destination host. Dumb write endpoint: verify+pwrite+ack per block. +// --------------------------------------------------------------------- + +func runSinkRole(path string, size, blockSize int64) error { + out := NewFrameWriter(os.Stdout) + in := NewFrameReader(os.Stdin) + + f, err := os.OpenFile(path, os.O_RDWR, 0) + if err != nil { + fmt.Fprintf(os.Stderr, "sink: open %s: %v\n", path, err) + return err + } + defer f.Close() + + if err := out.WriteFrame(frameReady, nil); err != nil { + return err + } + return runDestLoop(destLoopParams{File: f, BlockSize: blockSize, In: in, AckOut: out}) +} + +// --------------------------------------------------------------------- +// source-stream role: one-shot process spawned (over ssh, in pull mode) on +// the source host. Reads the hash table, then performs the same +// read/hash/compare/send loop a local push driver would, writing straight +// to its own stdout. +// --------------------------------------------------------------------- + +func runSourceStreamRole(path string, size, blockSize int64) error { + out := NewFrameWriter(os.Stdout) + in := NewFrameReader(os.Stdin) + + f, err := os.Open(path) + if err != nil { + fmt.Fprintf(os.Stderr, "source-stream: open %s: %v\n", path, err) + return err + } + defer f.Close() + + if err := out.WriteFrame(frameReady, nil); err != nil { + return err + } + + typ, payload, err := in.ReadFrame() + if err != nil { + return fmt.Errorf("source-stream: read hash table: %w", err) + } + if typ != frameHashTable { + return fmt.Errorf("source-stream: expected hash table frame, got type %d", typ) + } + hashes, err := unflattenHashes(payload) + if err != nil { + return err + } + + var copied, skipped 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: int64(len(hashes))}) + } + + return runSourceLoop(sourceLoopParams{ + File: f, Size: size, BlockSize: blockSize, Hashes: hashes, Out: out, + OnSkip: func(uint64) { skipped++; maybeProgress() }, + OnSend: func(uint64, [32]byte) { copied++; maybeProgress() }, + }) +} + +// --------------------------------------------------------------------- +// resultBatcher: coalesces BlockResult entries into occasional +// block_done_batch messages so the manager isn't hit with one control +// message per synced block. +// --------------------------------------------------------------------- + +type resultBatcher struct { + out *FrameWriter + mu sync.Mutex + buf []BlockResult + last time.Time +} + +func newResultBatcher(out *FrameWriter) *resultBatcher { + return &resultBatcher{out: out, last: time.Now()} +} + +func (b *resultBatcher) add(r BlockResult) { + b.mu.Lock() + b.buf = append(b.buf, r) + shouldFlush := len(b.buf) >= 256 || time.Since(b.last) > 500*time.Millisecond + b.mu.Unlock() + if shouldFlush { + b.flush() + } +} + +func (b *resultBatcher) flush() { + b.mu.Lock() + if len(b.buf) == 0 { + b.mu.Unlock() + return + } + entries := b.buf + b.buf = nil + b.last = time.Now() + b.mu.Unlock() + _ = b.out.WriteJSON(CtrlMsg{Type: msgBlockDoneBatch, Entries: entries}) +} diff --git a/control.go b/control.go new file mode 100644 index 0000000..8e89a55 --- /dev/null +++ b/control.go @@ -0,0 +1,204 @@ +package main + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" +) + +// Controller is the manager's handle on one long-lived control agent +// (spawned locally or over ssh), used for stat/prepare/connect_push/ +// connect_pull/close. The manager never does any source/dest I/O itself — +// every byte of the file/device it's syncing is read or written by an +// agent process, either this one or the sink/source-stream it spawns. +type Controller struct { + tag string + cmd *exec.Cmd + in io.WriteCloser + fw *FrameWriter + fr *FrameReader +} + +func startController(spec Spec, tag string, cfg *SyncConfig) (*Controller, error) { + var cmd *exec.Cmd + if spec.IsLocal() { + cmd = localAgentCommand([]string{"agent", "--role", "control"}) + } else { + remoteArgs := []string{cfg.RemoteBin, "agent", "--role", "control"} + cmd = sshCommand(cfg.SSHBin, cfg.SSHOpts, false, cfg.ConnectTimeoutSec, spec.User, spec.Host, remoteArgs) + } + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, err + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, err + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, err + } + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("start %s control agent: %w", tag, err) + } + go relayPrefixed(stderr, tag) + + return &Controller{tag: tag, cmd: cmd, in: stdin, fw: NewFrameWriter(stdin), fr: NewFrameReader(stdout)}, nil +} + +func relayPrefixed(r io.Reader, tag string) { + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 64*1024), 1024*1024) + for sc.Scan() { + fmt.Fprintf(os.Stderr, "[%s] %s\n", tag, sc.Text()) + } +} + +func (c *Controller) readOne() (CtrlMsg, error) { + typ, payload, err := c.fr.ReadFrame() + if err != nil { + return CtrlMsg{}, fmt.Errorf("%s: read control frame: %w", c.tag, err) + } + if typ != frameCtrlJSON { + return CtrlMsg{}, fmt.Errorf("%s: unexpected frame type %d on control channel", c.tag, typ) + } + var m CtrlMsg + if err := json.Unmarshal(payload, &m); err != nil { + return CtrlMsg{}, fmt.Errorf("%s: decode control message: %w", c.tag, err) + } + return m, nil +} + +func (c *Controller) call(req CtrlMsg) (CtrlMsg, error) { + if err := c.fw.WriteJSON(req); err != nil { + return CtrlMsg{}, fmt.Errorf("%s: send %s: %w", c.tag, req.Type, err) + } + return c.readOne() +} + +func (c *Controller) Stat(path string) (PathInfo, error) { + resp, err := c.call(CtrlMsg{Type: msgStat, Path: path}) + if err != nil { + return PathInfo{}, err + } + switch resp.Type { + case msgStatOK: + return PathInfo{Exists: resp.Exists, IsDevice: resp.IsDevice, Size: resp.Size}, nil + case msgError: + return PathInfo{}, fmt.Errorf("%s: stat %s: %s", c.tag, path, resp.Message) + default: + return PathInfo{}, fmt.Errorf("%s: unexpected response %q to stat", c.tag, resp.Type) + } +} + +func (c *Controller) Prepare(path string, size int64) error { + resp, err := c.call(CtrlMsg{Type: msgPrepare, Path: path, Size: size}) + if err != nil { + return err + } + switch resp.Type { + case msgPrepareOK: + return nil + case msgError: + return fmt.Errorf("%s: prepare %s: %s", c.tag, path, resp.Message) + default: + return fmt.Errorf("%s: unexpected response %q to prepare", c.tag, resp.Type) + } +} + +// transferCallbacks receives streaming updates while a connect_push or +// connect_pull is in flight. +type transferCallbacks struct { + onProgress func(CtrlMsg) + onBlockDone func(BlockResult) +} + +// ConnectPush asks the source control agent to try connecting straight out +// to the destination host and driving the whole transfer itself. ok=false +// with a non-empty reason means the SSH handshake didn't succeed (caller +// should try ConnectPull instead); a non-nil err means something failed +// after the transfer was already committed. +func (c *Controller) ConnectPush(req CtrlMsg, hashes [][32]byte, cb transferCallbacks) (ok bool, reason string, err error) { + req.Type = msgConnectPush + return c.connectAndPump(req, hashes, cb, msgPushOK, msgPushFailed) +} + +// ConnectPull asks the destination control agent to try connecting out to +// the source host and pulling the transfer. +func (c *Controller) ConnectPull(req CtrlMsg, hashes [][32]byte, cb transferCallbacks) (ok bool, reason string, err error) { + req.Type = msgConnectPull + return c.connectAndPump(req, hashes, cb, msgPullOK, msgPullFailed) +} + +func (c *Controller) connectAndPump(req CtrlMsg, hashes [][32]byte, cb transferCallbacks, okType, failedType string) (bool, string, error) { + if err := c.fw.WriteJSON(req); err != nil { + return false, "", fmt.Errorf("%s: send %s: %w", c.tag, req.Type, err) + } + if err := c.fw.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil { + return false, "", fmt.Errorf("%s: send hash table: %w", c.tag, err) + } + for { + m, err := c.readOne() + if err != nil { + return false, "", err + } + switch m.Type { + case msgProgress: + if cb.onProgress != nil { + cb.onProgress(m) + } + case msgBlockDoneBatch: + if cb.onBlockDone != nil { + for _, e := range m.Entries { + cb.onBlockDone(e) + } + } + case okType: + return true, "", nil + case failedType: + return false, m.Reason, nil + case msgError: + return false, "", fmt.Errorf("%s: %s", c.tag, m.Message) + default: + return false, "", fmt.Errorf("%s: unexpected message %q during transfer", c.tag, m.Type) + } + } +} + +func (c *Controller) Close() error { + _ = c.fw.WriteJSON(CtrlMsg{Type: msgClose}) + _ = c.in.Close() + err := c.cmd.Wait() + if err != nil { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { + return nil // agent exiting after we closed its stdin is expected + } + return err + } + return nil +} + +func flattenHashes(hashes [][32]byte) []byte { + buf := make([]byte, len(hashes)*32) + for i, h := range hashes { + copy(buf[i*32:], h[:]) + } + return buf +} + +func unflattenHashes(b []byte) ([][32]byte, error) { + if len(b)%32 != 0 { + return nil, fmt.Errorf("hash table: %d bytes is not a multiple of 32", len(b)) + } + out := make([][32]byte, len(b)/32) + for i := range out { + copy(out[i][:], b[i*32:i*32+32]) + } + return out, nil +} diff --git a/ctrlmsg.go b/ctrlmsg.go new file mode 100644 index 0000000..9b04f04 --- /dev/null +++ b/ctrlmsg.go @@ -0,0 +1,71 @@ +package main + +// CtrlMsg is the single envelope used for every message exchanged between +// the manager and a control agent over frameCtrlJSON frames. It is +// deliberately denormalized (not one struct per message type) to keep +// (de)serialization trivial. +type CtrlMsg struct { + Type string `json:"type"` + + // stat request/response + Path string `json:"path,omitempty"` + Exists bool `json:"exists,omitempty"` + IsDevice bool `json:"isDevice,omitempty"` + Size int64 `json:"size,omitempty"` + + // 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"` + // 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 + // over ssh — no loopback SSH access required for same-machine jobs. + PeerLocal bool `json:"peerLocal,omitempty"` + RemoteBin string `json:"remoteBin,omitempty"` + SSHBin string `json:"sshBin,omitempty"` + SSHOpts []string `json:"sshOpts,omitempty"` + ConnectTimeoutSec int `json:"connectTimeoutSec,omitempty"` + BlockSize int64 `json:"blockSize,omitempty"` + + // failure/error detail + Reason string `json:"reason,omitempty"` + Message string `json:"message,omitempty"` + + // progress / block_done_batch + Copied int64 `json:"copied,omitempty"` + Skipped int64 `json:"skipped,omitempty"` + TotalBlocks int64 `json:"totalBlocks,omitempty"` + BytesCopied int64 `json:"bytesCopied,omitempty"` + Entries []BlockResult `json:"entries,omitempty"` + + // log + Level string `json:"level,omitempty"` +} + +// BlockResult is one confirmed-written block's new hash, reported from a +// control agent back to the manager so it can update the job state file. +type BlockResult struct { + Index uint64 `json:"i"` + Hash string `json:"h"` // hex-encoded 32 bytes +} + +const ( + msgStat = "stat" + msgStatOK = "stat_ok" + msgPrepare = "prepare" + msgPrepareOK = "prepare_ok" + msgConnectPush = "connect_push" + msgConnectPull = "connect_pull" + msgPushOK = "push_ok" + msgPushFailed = "push_failed" + msgPullOK = "pull_ok" + msgPullFailed = "pull_failed" + msgProgress = "progress" + msgBlockDoneBatch = "block_done_batch" + msgLog = "log" + msgError = "error" + msgClose = "close" + msgBye = "bye" +) diff --git a/device.go b/device.go new file mode 100644 index 0000000..068cdd0 --- /dev/null +++ b/device.go @@ -0,0 +1,62 @@ +package main + +import ( + "fmt" + "os" +) + +// PathInfo describes a source/dest path as seen locally by whichever +// process (control agent, sink, source-stream) actually opens it. +type PathInfo struct { + Exists bool + IsDevice bool + Size int64 +} + +func statPath(path string) (PathInfo, error) { + fi, err := os.Stat(path) + if err != nil { + if os.IsNotExist(err) { + return PathInfo{Exists: false}, nil + } + return PathInfo{}, err + } + mode := fi.Mode() + isDevice := mode&os.ModeDevice != 0 && mode&os.ModeCharDevice == 0 + size := fi.Size() + if isDevice { + sz, err := blockDeviceSize(path) + if err != nil { + return PathInfo{}, fmt.Errorf("stat block device %s: %w", path, err) + } + size = sz + } + return PathInfo{Exists: true, IsDevice: isDevice, Size: size}, nil +} + +// prepareDest makes the destination ready to receive exactly targetSize +// bytes: a regular file is created if missing and truncated (grown or +// shrunk) to targetSize; a block device is only validated to be large +// enough (it can't be resized) — the caller is expected to have already +// capped targetSize at the device's own size when it is the limiting side. +func prepareDest(path string, targetSize int64) error { + info, err := statPath(path) + if err != nil { + return err + } + if info.Exists && info.IsDevice { + if targetSize > info.Size { + return fmt.Errorf("destination device %s is only %d bytes, need %d", path, info.Size, targetSize) + } + return nil + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644) + if err != nil { + return fmt.Errorf("open/create %s: %w", path, err) + } + defer f.Close() + if err := f.Truncate(targetSize); err != nil { + return fmt.Errorf("truncate %s to %d: %w", path, targetSize, err) + } + return nil +} diff --git a/device_linux.go b/device_linux.go new file mode 100644 index 0000000..ee40272 --- /dev/null +++ b/device_linux.go @@ -0,0 +1,28 @@ +//go:build linux + +package main + +import ( + "os" + "unsafe" + + "syscall" +) + +// BLKGETSIZE64 = _IOR(0x12, 114, sizeof(uint64)) on Linux. +const blkGetSize64 = 0x80081272 + +func blockDeviceSize(path string) (int64, error) { + f, err := os.Open(path) + if err != nil { + return 0, err + } + defer f.Close() + + var size uint64 + _, _, errno := syscall.Syscall(syscall.SYS_IOCTL, f.Fd(), blkGetSize64, uintptr(unsafe.Pointer(&size))) + if errno != 0 { + return 0, errno + } + return int64(size), nil +} diff --git a/device_other.go b/device_other.go new file mode 100644 index 0000000..ae13f7e --- /dev/null +++ b/device_other.go @@ -0,0 +1,9 @@ +//go:build !linux + +package main + +import "fmt" + +func blockDeviceSize(path string) (int64, error) { + return 0, fmt.Errorf("block device size detection is only implemented on linux (got path %s)", path) +} diff --git a/frame.go b/frame.go new file mode 100644 index 0000000..fcef468 --- /dev/null +++ b/frame.go @@ -0,0 +1,136 @@ +package main + +import ( + "bufio" + "encoding/binary" + "encoding/json" + "fmt" + "io" + "sync" +) + +// Wire format shared by the control channel (manager <-> control agent) and +// the data channel (source-agent/dest-agent <-> sink/source-stream). Every +// frame is: +// +// uint32 BE length (= 1 + len(payload)) +// byte type +// []byte payload (length-1 bytes) +const maxFrameLen = 256 * 1024 * 1024 // sanity cap, well above block-size + header defaults + +type frameType byte + +const ( + frameCtrlJSON frameType = 0x01 // payload: JSON object (control channel) + frameHashTable frameType = 0x02 // payload: raw concatenated 32-byte hashes + frameData frameType = 0x03 // payload: 8-byte index + 32-byte hash + block bytes + frameAck frameType = 0x04 // payload: 8-byte index + frameErr frameType = 0x05 // payload: 8-byte index + UTF-8 message + frameDone frameType = 0x06 // payload: empty + frameReady frameType = 0x07 // payload: empty +) + +// FrameWriter serializes concurrent writers onto one underlying stream. +type FrameWriter struct { + mu sync.Mutex + w io.Writer +} + +func NewFrameWriter(w io.Writer) *FrameWriter { return &FrameWriter{w: w} } + +func (fw *FrameWriter) WriteFrame(typ frameType, payload []byte) error { + fw.mu.Lock() + defer fw.mu.Unlock() + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(1+len(payload))) + if _, err := fw.w.Write(hdr[:]); err != nil { + return err + } + if _, err := fw.w.Write([]byte{byte(typ)}); err != nil { + return err + } + if len(payload) > 0 { + if _, err := fw.w.Write(payload); err != nil { + return err + } + } + return nil +} + +func (fw *FrameWriter) WriteJSON(v any) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + return fw.WriteFrame(frameCtrlJSON, b) +} + +// FrameReader reads frames from a stream. Not safe for concurrent use. +type FrameReader struct { + r *bufio.Reader +} + +func NewFrameReader(r io.Reader) *FrameReader { + return &FrameReader{r: bufio.NewReaderSize(r, 64*1024)} +} + +func (fr *FrameReader) ReadFrame() (frameType, []byte, error) { + var hdr [4]byte + if _, err := io.ReadFull(fr.r, hdr[:]); err != nil { + return 0, nil, err + } + n := binary.BigEndian.Uint32(hdr[:]) + if n == 0 || n > maxFrameLen { + return 0, nil, fmt.Errorf("frame: invalid length %d", n) + } + buf := make([]byte, n) + if _, err := io.ReadFull(fr.r, buf); err != nil { + return 0, nil, err + } + return frameType(buf[0]), buf[1:], nil +} + +func encodeDataFrame(index uint64, hash [32]byte, payload []byte) []byte { + buf := make([]byte, 8+32+len(payload)) + binary.BigEndian.PutUint64(buf[0:8], index) + copy(buf[8:40], hash[:]) + copy(buf[40:], payload) + return buf +} + +func decodeDataFrame(b []byte) (index uint64, hash [32]byte, payload []byte, err error) { + if len(b) < 40 { + return 0, hash, nil, fmt.Errorf("data frame too short: %d bytes", len(b)) + } + index = binary.BigEndian.Uint64(b[0:8]) + copy(hash[:], b[8:40]) + payload = b[40:] + return index, hash, payload, nil +} + +func encodeIndexFrame(index uint64) []byte { + buf := make([]byte, 8) + binary.BigEndian.PutUint64(buf, index) + return buf +} + +func decodeIndexFrame(b []byte) (uint64, error) { + if len(b) < 8 { + return 0, fmt.Errorf("index frame too short: %d bytes", len(b)) + } + return binary.BigEndian.Uint64(b[:8]), nil +} + +func encodeErrFrame(index uint64, msg string) []byte { + buf := make([]byte, 8+len(msg)) + binary.BigEndian.PutUint64(buf[0:8], index) + copy(buf[8:], msg) + return buf +} + +func decodeErrFrame(b []byte) (uint64, string, error) { + if len(b) < 8 { + return 0, "", fmt.Errorf("err frame too short: %d bytes", len(b)) + } + return binary.BigEndian.Uint64(b[:8]), string(b[8:]), nil +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..ef15563 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module clonetool + +go 1.27.1 diff --git a/jobstate.go b/jobstate.go new file mode 100644 index 0000000..592f446 --- /dev/null +++ b/jobstate.go @@ -0,0 +1,204 @@ +package main + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "os" + "path/filepath" +) + +const jobStateMagic = "CTJ1" + +// JobState is the manager-local trust anchor for one named job: the +// per-block hashes of the destination as of the last confirmed write. It is +// never derived by reading the destination — only by recording what this +// tool itself wrote there (or confirmed already matched). +type JobState struct { + BlockSize int64 + SrcSpec string + DstSpec string + Size int64 + Hashes [][32]byte // zero hash = never synced +} + +func statePathFor(stateDir, job string) string { + return filepath.Join(stateDir, job+".state") +} + +func defaultStateDir() string { + home, err := os.UserHomeDir() + if err != nil { + return ".clonetool/jobs" + } + return filepath.Join(home, ".clonetool", "jobs") +} + +// LoadJobState reads the job's state file. If it doesn't exist yet, an +// empty state is returned (first run). If it exists but was recorded +// against a different source/dest spec, an error is returned unless force +// is set, in which case the old hash history is discarded. +func LoadJobState(path, srcSpec, dstSpec string, blockSize int64, force bool) (*JobState, error) { + f, err := os.Open(path) + if err != nil { + if os.IsNotExist(err) { + return &JobState{BlockSize: blockSize, SrcSpec: srcSpec, DstSpec: dstSpec}, nil + } + return nil, err + } + defer f.Close() + + r := bufio.NewReader(f) + magic := make([]byte, 4) + if _, err := io.ReadFull(r, magic); err != nil { + return nil, fmt.Errorf("read job state %s: %w", path, err) + } + if string(magic) != jobStateMagic { + return nil, fmt.Errorf("job state %s: bad magic", path) + } + var version uint8 + if err := binary.Read(r, binary.BigEndian, &version); err != nil { + return nil, err + } + if version != 1 { + return nil, fmt.Errorf("job state %s: unsupported version %d", path, version) + } + var storedBlockSize uint32 + if err := binary.Read(r, binary.BigEndian, &storedBlockSize); err != nil { + return nil, err + } + var blockCount uint64 + if err := binary.Read(r, binary.BigEndian, &blockCount); err != nil { + return nil, err + } + storedSrc, err := readLPString(r) + if err != nil { + return nil, err + } + storedDst, err := readLPString(r) + if err != nil { + return nil, err + } + var size int64 + if err := binary.Read(r, binary.BigEndian, &size); err != nil { + return nil, err + } + + if !force && (storedSrc != srcSpec || storedDst != dstSpec) { + return nil, fmt.Errorf( + "job state %s was recorded for source=%q dest=%q, but this run uses source=%q dest=%q; "+ + "reusing its hash history against different endpoints could skip blocks never written there. "+ + "Pass --force to rebind this job (discards hash history)", + path, storedSrc, storedDst, srcSpec, dstSpec) + } + + hashes := make([][32]byte, blockCount) + for i := range hashes { + if _, err := io.ReadFull(r, hashes[i][:]); err != nil { + return nil, fmt.Errorf("read job state %s: %w", path, err) + } + } + + js := &JobState{BlockSize: int64(storedBlockSize), SrcSpec: srcSpec, DstSpec: dstSpec, Size: size, Hashes: hashes} + if force { + js.SrcSpec, js.DstSpec = srcSpec, dstSpec + } + if js.BlockSize != blockSize { + // Block size changed: the stored per-index hashes no longer line up + // with block boundaries, so start over rather than misinterpret them. + js.BlockSize = blockSize + js.Hashes = nil + } + return js, nil +} + +// Resize grows (with zero/"unknown" hashes) or shrinks the hash table to +// match a new block count. +func (js *JobState) Resize(blockCount uint64) { + if uint64(len(js.Hashes)) == blockCount { + return + } + grown := make([][32]byte, blockCount) + copy(grown, js.Hashes) + js.Hashes = grown +} + +func (js *JobState) Save(path string) error { + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + return err + } + tmp := path + ".tmp" + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644) + if err != nil { + return err + } + w := bufio.NewWriter(f) + writeErr := func() error { + if _, err := w.WriteString(jobStateMagic); err != nil { + return err + } + if err := binary.Write(w, binary.BigEndian, uint8(1)); err != nil { + return err + } + if err := binary.Write(w, binary.BigEndian, uint32(js.BlockSize)); err != nil { + return err + } + if err := binary.Write(w, binary.BigEndian, uint64(len(js.Hashes))); err != nil { + return err + } + if err := writeLPString(w, js.SrcSpec); err != nil { + return err + } + if err := writeLPString(w, js.DstSpec); err != nil { + return err + } + if err := binary.Write(w, binary.BigEndian, js.Size); err != nil { + return err + } + for _, h := range js.Hashes { + if _, err := w.Write(h[:]); err != nil { + return err + } + } + return w.Flush() + }() + if writeErr != nil { + f.Close() + os.Remove(tmp) + return writeErr + } + if err := f.Sync(); err != nil { + f.Close() + os.Remove(tmp) + return err + } + if err := f.Close(); err != nil { + os.Remove(tmp) + return err + } + return os.Rename(tmp, path) +} + +func readLPString(r io.Reader) (string, error) { + var n uint16 + if err := binary.Read(r, binary.BigEndian, &n); err != nil { + return "", err + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return "", err + } + return string(buf), nil +} + +func writeLPString(w io.Writer, s string) error { + if len(s) > 65535 { + return fmt.Errorf("string too long to store (%d bytes)", len(s)) + } + if err := binary.Write(w, binary.BigEndian, uint16(len(s))); err != nil { + return err + } + _, err := io.WriteString(w, s) + return err +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..9482f79 --- /dev/null +++ b/main.go @@ -0,0 +1,132 @@ +package main + +import ( + "flag" + "fmt" + "os" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + + var err error + switch os.Args[1] { + case "sync": + err = cmdSync(os.Args[2:]) + case "agent": + err = cmdAgent(os.Args[2:]) + case "-h", "--help", "help": + usage() + return + default: + usage() + os.Exit(2) + } + if err != nil { + fmt.Fprintln(os.Stderr, "clonetool: "+err.Error()) + os.Exit(1) + } +} + +func usage() { + fmt.Fprint(os.Stderr, `clonetool - block-level file/device sync + +Usage: + clonetool sync --job NAME --source LOC --dest LOC [options] + clonetool agent --role {control|sink|source-stream} ... (internal, spawned automatically) + +LOC is either a local path, or [user@]host:path for a path reached over SSH. +Source, destination, and the machine running "sync" (the manager) may all be +different machines: the manager only orchestrates, it never reads or writes +a single block itself. + +Options for sync: + --block-size SIZE block size, e.g. 4M (default 4M) + --state-dir DIR where job hash-tables are stored (default ~/.clonetool/jobs) + --yes don't prompt before shrinking an existing destination file + --force rebind this job to a new source/dest, discarding hash history + --connect-timeout SEC ssh connect timeout for the push/pull direction probe (default 8) + --ssh PATH ssh binary to use (default "ssh") + --ssh-opt OPT extra "-o OPT" passed to ssh (repeatable) + --remote-bin PATH path to clonetool on remote hosts (default "clonetool") + --manager-host HOST address peers should use to reach this machine, when + source or dest has no host part (defaults to the local hostname) + +The destination must not be modified by anything else between syncs of the +same job: repeated runs trust the job's recorded hash table instead of +re-reading the destination. +`) +} + +type stringSlice []string + +func (s *stringSlice) String() string { return fmt.Sprint([]string(*s)) } +func (s *stringSlice) Set(v string) error { + *s = append(*s, v) + return nil +} + +func cmdSync(args []string) error { + fs := flag.NewFlagSet("sync", flag.ContinueOnError) + job := fs.String("job", "", "job name (required)") + source := fs.String("source", "", "source location (required)") + dest := fs.String("dest", "", "destination location (required)") + blockSizeStr := fs.String("block-size", "4M", "block size, e.g. 4M") + stateDir := fs.String("state-dir", defaultStateDir(), "job state directory") + yes := fs.Bool("yes", false, "don't prompt before shrinking destination") + force := fs.Bool("force", false, "rebind job to a new source/dest") + connectTimeout := fs.Int("connect-timeout", defaultConnectTimeoutSec, "ssh connect timeout (seconds)") + sshBin := fs.String("ssh", "ssh", "ssh binary") + remoteBin := fs.String("remote-bin", "clonetool", "clonetool path on remote hosts") + managerHost := fs.String("manager-host", "", "address peers use to reach this machine") + var sshOpts stringSlice + fs.Var(&sshOpts, "ssh-opt", `extra "-o OPT" passed to ssh (repeatable)`) + if err := fs.Parse(args); err != nil { + return err + } + + if *job == "" || *source == "" || *dest == "" { + fs.Usage() + return fmt.Errorf("--job, --source and --dest are required") + } + blockSize, err := parseSize(*blockSizeStr) + if err != nil { + return fmt.Errorf("--block-size: %w", err) + } + + return runSync(SyncConfig{ + Job: *job, Source: *source, Dest: *dest, BlockSize: blockSize, StateDir: *stateDir, + Yes: *yes, Force: *force, ConnectTimeoutSec: *connectTimeout, SSHBin: *sshBin, + SSHOpts: sshOpts, RemoteBin: *remoteBin, ManagerHost: *managerHost, + }) +} + +func parseSize(s string) (int64, error) { + if s == "" { + return 0, fmt.Errorf("empty size") + } + mult := int64(1) + numPart := s + switch s[len(s)-1] { + case 'k', 'K': + mult = 1024 + numPart = s[:len(s)-1] + case 'm', 'M': + mult = 1024 * 1024 + numPart = s[:len(s)-1] + case 'g', 'G': + mult = 1024 * 1024 * 1024 + numPart = s[:len(s)-1] + } + var n int64 + if _, err := fmt.Sscanf(numPart, "%d", &n); err != nil { + return 0, fmt.Errorf("invalid size %q", s) + } + if n <= 0 { + return 0, fmt.Errorf("size must be positive") + } + return n * mult, nil +} diff --git a/manager.go b/manager.go new file mode 100644 index 0000000..48b7c03 --- /dev/null +++ b/manager.go @@ -0,0 +1,212 @@ +package main + +import ( + "bufio" + "encoding/hex" + "fmt" + "os" + "strings" + "time" +) + +const defaultBlockSize = 4 * 1024 * 1024 +const defaultConnectTimeoutSec = 8 + +// SyncConfig holds everything the `sync` command needs. It is also handed +// (the transport-relevant fields of it) to control agents as part of +// connect_push/connect_pull requests, so they know how to reach the peer. +type SyncConfig struct { + Job string + Source string + Dest string + BlockSize int64 + StateDir string + Yes bool + Force bool + ConnectTimeoutSec int + SSHBin string + SSHOpts []string + RemoteBin string + ManagerHost string +} + +func runSync(cfg SyncConfig) error { + srcSpec, err := parseSpec(cfg.Source) + if err != nil { + return fmt.Errorf("--source: %w", err) + } + dstSpec, err := parseSpec(cfg.Dest) + if err != nil { + return fmt.Errorf("--dest: %w", err) + } + if err := checkNotSame(srcSpec, dstSpec); err != nil { + return err + } + + srcCtrl, err := startController(srcSpec, "source", &cfg) + if err != nil { + return fmt.Errorf("start source control agent: %w", err) + } + defer srcCtrl.Close() + + dstCtrl, err := startController(dstSpec, "dest", &cfg) + if err != nil { + return fmt.Errorf("start dest control agent: %w", err) + } + defer dstCtrl.Close() + + srcInfo, err := srcCtrl.Stat(srcSpec.Path) + if err != nil { + return fmt.Errorf("stat source: %w", err) + } + if !srcInfo.Exists { + return fmt.Errorf("source %s does not exist", srcSpec) + } + + dstInfo, err := dstCtrl.Stat(dstSpec.Path) + if err != nil { + return fmt.Errorf("stat destination: %w", err) + } + + targetSize, err := computeTargetSize(srcSpec, srcInfo, dstSpec, dstInfo, cfg.Yes) + if err != nil { + return err + } + + if err := dstCtrl.Prepare(dstSpec.Path, targetSize); err != nil { + return fmt.Errorf("prepare destination: %w", err) + } + + blockCount := (targetSize + cfg.BlockSize - 1) / cfg.BlockSize + statePath := statePathFor(cfg.StateDir, cfg.Job) + state, err := LoadJobState(statePath, cfg.Source, cfg.Dest, cfg.BlockSize, cfg.Force) + if err != nil { + return err + } + state.Resize(uint64(blockCount)) + + lastSave := time.Now() + sinceSave := 0 + checkpoint := func(force bool) { + sinceSave++ + if !force && sinceSave < 2000 && time.Since(lastSave) < 5*time.Second { + return + } + if err := state.Save(statePath); err != nil { + fmt.Fprintf(os.Stderr, "warning: could not checkpoint job state: %v\n", err) + return + } + lastSave = time.Now() + sinceSave = 0 + } + + cb := transferCallbacks{ + onProgress: func(m CtrlMsg) { printProgress(m) }, + onBlockDone: func(r BlockResult) { + h, err := hex.DecodeString(r.Hash) + if err != nil || len(h) != 32 { + return + } + if r.Index < uint64(len(state.Hashes)) { + copy(state.Hashes[r.Index][:], h) + } + checkpoint(false) + }, + } + + bothLocal := srcSpec.IsLocal() && dstSpec.IsLocal() + srcHost, srcUser := resolveConnectHost(srcSpec, &cfg) + dstHost, dstUser := resolveConnectHost(dstSpec, &cfg) + + pushReq := CtrlMsg{ + Path: srcSpec.Path, Size: targetSize, BlockSize: cfg.BlockSize, + PeerHost: dstHost, PeerUser: dstUser, PeerPath: dstSpec.Path, PeerLocal: bothLocal, + RemoteBin: cfg.RemoteBin, SSHBin: cfg.SSHBin, SSHOpts: cfg.SSHOpts, ConnectTimeoutSec: cfg.ConnectTimeoutSec, + } + if bothLocal { + fmt.Fprintf(os.Stderr, "both source and dest are local; syncing directly (no ssh) ...\n") + } else { + fmt.Fprintf(os.Stderr, "attempting push %s -> %s ...\n", srcSpec, dstSpec) + } + ok, reason, err := srcCtrl.ConnectPush(pushReq, state.Hashes, cb) + if err != nil { + return err + } + if !ok { + fmt.Fprintf(os.Stderr, "push not possible (%s); trying pull %s <- %s ...\n", reason, dstSpec, srcSpec) + pullReq := CtrlMsg{ + Path: dstSpec.Path, Size: targetSize, BlockSize: cfg.BlockSize, + PeerHost: srcHost, PeerUser: srcUser, PeerPath: srcSpec.Path, PeerLocal: bothLocal, + RemoteBin: cfg.RemoteBin, SSHBin: cfg.SSHBin, SSHOpts: cfg.SSHOpts, ConnectTimeoutSec: cfg.ConnectTimeoutSec, + } + ok2, reason2, err2 := dstCtrl.ConnectPull(pullReq, state.Hashes, cb) + if err2 != nil { + return err2 + } + if !ok2 { + return fmt.Errorf( + "could not establish a direct connection in either direction (push: %s; pull: %s); "+ + "run the manager on the source or destination host, or set up SSH connectivity in at least one direction", + reason, reason2) + } + } + + fmt.Fprintln(os.Stderr) + state.Size = targetSize + checkpoint(true) + fmt.Fprintf(os.Stderr, "done: job %q, %d blocks, source=%s dest=%s\n", cfg.Job, blockCount, srcSpec, dstSpec) + return nil +} + +func resolveConnectHost(spec Spec, cfg *SyncConfig) (host, user string) { + if !spec.IsLocal() { + return spec.Host, spec.User + } + host = cfg.ManagerHost + if host == "" { + if h, err := os.Hostname(); err == nil { + host = h + } + } + return host, "" +} + +// computeTargetSize applies the sizing/truncation rules: a block-device +// destination can't grow, so the source must fit inside it (sync exactly +// min(src,dst), leaving the remainder of the device untouched); a +// regular-file destination is truncated to the source's size, shrinking +// with confirmation if it currently holds more data than that. +func computeTargetSize(srcSpec Spec, srcInfo PathInfo, dstSpec Spec, dstInfo PathInfo, yes bool) (int64, error) { + if dstInfo.Exists && dstInfo.IsDevice { + if srcInfo.Size > dstInfo.Size { + return 0, fmt.Errorf("source %s (%s) is larger than destination device %s (%s); a device can't be grown", + srcSpec, humanBytes(srcInfo.Size), dstSpec, humanBytes(dstInfo.Size)) + } + return srcInfo.Size, nil + } + if dstInfo.Exists && dstInfo.Size > srcInfo.Size { + if !yes { + if !confirmShrink(dstSpec, dstInfo.Size, srcInfo.Size) { + return 0, fmt.Errorf("aborted: destination %s would be truncated from %s to %s", dstSpec, humanBytes(dstInfo.Size), humanBytes(srcInfo.Size)) + } + } + } + return srcInfo.Size, nil +} + +func confirmShrink(dstSpec Spec, oldSize, newSize int64) bool { + fmt.Fprintf(os.Stderr, "destination %s is %s; it will be truncated to %s to match the source. Continue? [y/N] ", + dstSpec, humanBytes(oldSize), humanBytes(newSize)) + reader := bufio.NewReader(os.Stdin) + line, _ := reader.ReadString('\n') + line = strings.TrimSpace(strings.ToLower(line)) + return line == "y" || line == "yes" +} + +func printProgress(m CtrlMsg) { + total := m.TotalBlocks + if total <= 0 { + total = m.Copied + m.Skipped + } + fmt.Fprintf(os.Stderr, "\rblocks: %d/%d copied=%d skipped=%d ", m.Copied+m.Skipped, total, m.Copied, m.Skipped) +} diff --git a/prompts.md b/prompts.md new file mode 100644 index 0000000..fd0efa9 --- /dev/null +++ b/prompts.md @@ -0,0 +1,17 @@ +# CloneTool + +i need a clone-tool. + +it should be run from a single binary without any dependencies (maybe on go?). +it should be able to clone a big file or a block device from one computer to another or on the same compuser from one device to another. +it should compare blocks and if they differ, copy it from source to destination. +to prevent double reading, keep the blocks in memory until diff was checked. +i want to start it from one computer which is the manager of the copy process, source and destination can be other computers. +blocks should be copied directly from source to destination. +i dont want to have any dependencies. +it is allowed to use ssh and dd if nessesary. +maybe it makes sense to use a copy job name and track block hashes in some file to compare them without reading from destination (which must not be changed between syncs) i want to be able to run the sync multiple time in one job. +if source is smaller than target, stop after end of source if destination is a device. +if destination is a file, truncate it to the size of source + +first try to connect from source to destination, of not possible, try to connect from destination to source. transfer all data via ssh diff --git a/spec.go b/spec.go new file mode 100644 index 0000000..e12a0a0 --- /dev/null +++ b/spec.go @@ -0,0 +1,76 @@ +package main + +import ( + "fmt" + "path/filepath" + "strings" +) + +// Spec is a parsed source/dest location: [user@]host:path, or a bare local +// path (Host == ""). +type Spec struct { + Raw string + User string + Host string + Path string +} + +func (s Spec) IsLocal() bool { return s.Host == "" } + +func (s Spec) String() string { + if s.IsLocal() { + return s.Path + } + if s.User != "" { + return fmt.Sprintf("%s@%s:%s", s.User, s.Host, s.Path) + } + return fmt.Sprintf("%s:%s", s.Host, s.Path) +} + +// parseSpec parses "[user@]host:path" or a local "path". A leading "/", +// "./" or "../", or the absence of any colon, is treated as a local path so +// that ordinary absolute/relative paths are never mistaken for a host spec. +func parseSpec(raw string) (Spec, error) { + if raw == "" { + return Spec{}, fmt.Errorf("empty location") + } + if strings.HasPrefix(raw, "/") || strings.HasPrefix(raw, "./") || strings.HasPrefix(raw, "../") || !strings.Contains(raw, ":") { + return Spec{Raw: raw, Path: raw}, nil + } + idx := strings.Index(raw, ":") + hostpart := raw[:idx] + path := raw[idx+1:] + if hostpart == "" || path == "" || strings.ContainsAny(hostpart, "/ ") { + return Spec{}, fmt.Errorf("cannot parse location %q (expected [user@]host:path or a local path)", raw) + } + user := "" + host := hostpart + if at := strings.Index(hostpart, "@"); at >= 0 { + user = hostpart[:at] + host = hostpart[at+1:] + } + if host == "" { + return Spec{}, fmt.Errorf("cannot parse location %q: empty host", raw) + } + return Spec{Raw: raw, User: user, Host: host, Path: path}, nil +} + +// checkNotSame does a best-effort local check that source and dest don't +// refer to the exact same path, to avoid an obviously destructive mistake. +// It cannot resolve whether two different remote hostnames are actually the +// same machine. +func checkNotSame(src, dst Spec) error { + if src.IsLocal() != dst.IsLocal() { + return nil + } + if !src.IsLocal() && !strings.EqualFold(src.Host, dst.Host) { + return nil + } + if !src.IsLocal() && src.User != dst.User { + return nil + } + if filepath.Clean(src.Path) == filepath.Clean(dst.Path) { + return fmt.Errorf("source and destination resolve to the same path (%s)", src.Path) + } + return nil +} diff --git a/syncside.go b/syncside.go new file mode 100644 index 0000000..d3a64a5 --- /dev/null +++ b/syncside.go @@ -0,0 +1,135 @@ +package main + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + "io" + "os" +) + +// readBlockAt reads exactly min(len(buf), totalSize-offset) bytes at offset, +// treating a fully-satisfied read as success even if the underlying +// implementation also reports io.EOF for it. +func readBlockAt(f *os.File, buf []byte, offset, totalSize int64) (int, error) { + remaining := totalSize - offset + if remaining <= 0 { + return 0, io.EOF + } + want := int64(len(buf)) + if remaining < want { + want = remaining + } + n, err := f.ReadAt(buf[:want], offset) + if err != nil && !(err == io.EOF && int64(n) == want) { + return n, err + } + return n, nil +} + +// sourceLoopParams drives the single read -> hash -> compare -> maybe-send +// pass over a source path. It is used both by the standalone +// "source-stream" role (writing to its own stdout) and by a source control +// agent's push driver (writing into a spawned ssh subprocess's stdin). +type sourceLoopParams struct { + File *os.File // already opened for reading by the caller + Size int64 + BlockSize int64 + Hashes [][32]byte // previous known hashes, len == block count + Out *FrameWriter + OnSkip func(index uint64) + OnSend func(index uint64, hash [32]byte) // called after the DATA frame is written +} + +func runSourceLoop(p sourceLoopParams) error { + f := p.File + buf := make([]byte, p.BlockSize) + blockCount := uint64(len(p.Hashes)) + for index := uint64(0); index < blockCount; index++ { + offset := int64(index) * p.BlockSize + n, err := readBlockAt(f, buf, offset, p.Size) + if err != nil { + return fmt.Errorf("read %s at block %d: %w", f.Name(), index, err) + } + hash := sha256.Sum256(buf[:n]) + if hash == p.Hashes[index] { + if p.OnSkip != nil { + p.OnSkip(index) + } + continue + } + if err := p.Out.WriteFrame(frameData, encodeDataFrame(index, hash, buf[:n])); err != nil { + return fmt.Errorf("send block %d: %w", index, err) + } + if p.OnSend != nil { + p.OnSend(index, hash) + } + } + return p.Out.WriteFrame(frameDone, nil) +} + +// destLoopParams drives the receive -> verify -> pwrite pass on the +// destination path. It is used both by the standalone "sink" role (reading +// from its own stdin, and required to ack/err back over AckOut) and by a +// dest control agent's pull driver (reading from a spawned ssh subprocess's +// stdout; no AckOut needed since the write is confirmed locally, in the +// same process, before OnWritten is called). +type destLoopParams struct { + File *os.File // already opened for read/write by the caller + BlockSize int64 + In *FrameReader + AckOut *FrameWriter // optional + OnWritten func(index uint64, hash [32]byte) + // OnCtrlMsg handles an interleaved frameCtrlJSON frame (used only in + // 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) +} + +func runDestLoop(p destLoopParams) error { + f := p.File + for { + typ, payload, err := p.In.ReadFrame() + if err != nil { + return fmt.Errorf("read frame: %w", err) + } + switch typ { + case frameDone: + return f.Sync() + case frameData: + index, hash, block, err := decodeDataFrame(payload) + if err != nil { + return err + } + if sha256.Sum256(block) != hash { + if p.AckOut != nil { + _ = p.AckOut.WriteFrame(frameErr, encodeErrFrame(index, "hash mismatch after transfer")) + } + return fmt.Errorf("block %d: hash mismatch after transfer", index) + } + if _, err := f.WriteAt(block, int64(index)*p.BlockSize); 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.AckOut != nil { + if err := p.AckOut.WriteFrame(frameAck, encodeIndexFrame(index)); err != nil { + return err + } + } + if p.OnWritten != nil { + p.OnWritten(index, hash) + } + case frameCtrlJSON: + if p.OnCtrlMsg != nil { + var m CtrlMsg + if err := json.Unmarshal(payload, &m); err == nil { + p.OnCtrlMsg(m) + } + } + default: + return fmt.Errorf("unexpected frame type %d on data channel", typ) + } + } +} diff --git a/transport.go b/transport.go new file mode 100644 index 0000000..6c00bff --- /dev/null +++ b/transport.go @@ -0,0 +1,57 @@ +package main + +import ( + "fmt" + "os" + "os/exec" + "strings" +) + +// shellQuote wraps s in single quotes so it survives the remote shell that +// OpenSSH hands its non-option arguments to (ssh joins them itself; it does +// not exec the remote command with a distinct argv the way os/exec does). +func shellQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" +} + +func shellJoin(args []string) string { + parts := make([]string, len(args)) + for i, a := range args { + parts[i] = shellQuote(a) + } + return strings.Join(parts, " ") +} + +// sshCommand builds an *exec.Cmd that runs remoteArgs on host via ssh. +// batchMode disables interactive prompts (used for the push/pull direction +// probe, so an unreachable/unauthenticated attempt fails fast instead of +// hanging); it is left off for the manager's own control connections so a +// password prompt still works when the user is present at a terminal. +func sshCommand(sshBin string, extraOpts []string, batchMode bool, connectTimeoutSec int, user, host string, remoteArgs []string) *exec.Cmd { + args := []string{"-o", "StrictHostKeyChecking=accept-new"} + if batchMode { + args = append(args, "-o", "BatchMode=yes") + } + if connectTimeoutSec > 0 { + args = append(args, "-o", fmt.Sprintf("ConnectTimeout=%d", connectTimeoutSec)) + } + for _, o := range extraOpts { + args = append(args, "-o", o) + } + userHost := host + if user != "" { + userHost = user + "@" + host + } + args = append(args, userHost, shellJoin(remoteArgs)) + return exec.Command(sshBin, args...) +} + +// localAgentCommand re-execs this same binary as an agent, for a spec with +// no host part. +func localAgentCommand(remoteArgs []string) *exec.Cmd { + self, err := os.Executable() + if err != nil { + self = os.Args[0] + } + return exec.Command(self, remoteArgs...) +} diff --git a/util.go b/util.go new file mode 100644 index 0000000..643bbed --- /dev/null +++ b/util.go @@ -0,0 +1,49 @@ +package main + +import ( + "fmt" + "sync" +) + +// 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. +type limitedBuffer struct { + mu sync.Mutex + buf []byte + maxLen int +} + +func newLimitedBuffer(maxLen int) *limitedBuffer { + return &limitedBuffer{maxLen: maxLen} +} + +func (b *limitedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.buf = append(b.buf, p...) + if len(b.buf) > b.maxLen { + b.buf = b.buf[len(b.buf)-b.maxLen:] + } + return len(p), nil +} + +func (b *limitedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return string(b.buf) +} + +func humanBytes(n int64) string { + const unit = 1024 + if n < unit { + return fmt.Sprintf("%d B", n) + } + div, exp := int64(unit), 0 + for m := n / unit; m >= unit; m /= unit { + div *= unit + exp++ + } + units := "KMGTPE" + return fmt.Sprintf("%.1f%ciB", float64(n)/float64(div), units[exp]) +}