From 9e5e0fa8312d342c0954048949bc7005cee356d4 Mon Sep 17 00:00:00 2001 From: Alexander Gabriel Date: Sat, 5 Sep 2026 23:02:29 +0200 Subject: [PATCH] first batch --- .gitignore | 1 + README.md | 103 +++++++++ agent.go | 557 ++++++++++++++++++++++++++++++++++++++++++++++++ control.go | 204 ++++++++++++++++++ ctrlmsg.go | 71 ++++++ device.go | 62 ++++++ device_linux.go | 28 +++ device_other.go | 9 + frame.go | 136 ++++++++++++ go.mod | 3 + jobstate.go | 204 ++++++++++++++++++ main.go | 132 ++++++++++++ manager.go | 212 ++++++++++++++++++ prompts.md | 17 ++ spec.go | 76 +++++++ syncside.go | 135 ++++++++++++ transport.go | 57 +++++ util.go | 49 +++++ 18 files changed, 2056 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 agent.go create mode 100644 control.go create mode 100644 ctrlmsg.go create mode 100644 device.go create mode 100644 device_linux.go create mode 100644 device_other.go create mode 100644 frame.go create mode 100644 go.mod create mode 100644 jobstate.go create mode 100644 main.go create mode 100644 manager.go create mode 100644 prompts.md create mode 100644 spec.go create mode 100644 syncside.go create mode 100644 transport.go create mode 100644 util.go 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]) +}