From 63601f929d7dc0a232d433a78d85c9653fdbd3ef Mon Sep 17 00:00:00 2001 From: Alexander Gabriel Date: Sun, 6 Sep 2026 00:20:21 +0200 Subject: [PATCH] no state-saving any more --- README.md | 58 ++++++++++----- agent.go | 195 +++++++++++++++++++++++------------------------- control.go | 68 ++++++++++------- ctrlmsg.go | 55 +++++++------- deploy.go | 109 +++++++++++++++++++++++++++ jobstate.go | 204 --------------------------------------------------- main.go | 42 +++++++---- manager.go | 143 +++++++++++++++++++++--------------- syncside.go | 22 +++++- transport.go | 22 ++++-- 10 files changed, 452 insertions(+), 466 deletions(-) create mode 100644 deploy.go delete mode 100644 jobstate.go diff --git a/README.md b/README.md index da611fb..db34766 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,12 @@ system `ssh` client for remote endpoints. A block is only re-read once: it's hashed from the buffer it was read into, and that same buffer is what gets sent on if it differs — never read twice. +- **No state is kept between runs.** Every sync re-reads and re-hashes the + destination's *current* content and compares the source against that, so + a re-run only moves the blocks that actually differ and nothing needs to + be trusted from a previous run. (The destination side does the + destination hashing; in push mode it streams that table to the source, + in pull mode the destination agent hashes locally.) - 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 @@ -29,14 +35,22 @@ system `ssh` client for remote endpoints. - 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). +- **Self-deploy:** if a remote endpoint has no runnable `clonetool` on + `PATH` (or wherever `--remote-bin` points), the manager streams *this* + binary to `~/.clonetool/bin/clonetool` on that host over the existing + SSH connection and uses it — no install, no root. Disable with + `--deploy=false`. If the copied binary won't execute there (wrong CPU + architecture) the error says so; build one for the remote's arch + (`CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build`) and put it on `PATH` + or pass `--remote-bin`. +- **sudo:** if reading or writing an endpoint that is a **block device** + fails with a permission error, `--sudo=auto` (the default) transparently + restarts that side's agent — and the helper it spawns on the peer — + under `sudo`. Locally that may prompt for a password on your terminal; + **on a remote host it uses `sudo -n`, so passwordless sudo (NOPASSWD) + must be configured there** (a password prompt can't work — the transfer + protocol owns the SSH stdout stream). `--sudo=always` elevates from the + start; `--sudo=never` never does. - Sizing rules: - Destination is a **block device**: it can't be resized, so if the source is larger the job fails; otherwise exactly `min(source, dest)` @@ -52,14 +66,15 @@ system `ssh` client for remote endpoints. 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). +Copy the resulting binary to the manager host. Source and destination +hosts get it automatically (see self-deploy above), or place it yourself +and point `--remote-bin` at it. The binary is architecture-specific — +cross-compile (`GOOS`/`GOARCH`) if your hosts differ. ## Usage ``` -clonetool sync --job NAME --source LOC --dest LOC [options] +clonetool sync --source LOC --dest LOC [options] ``` `LOC` is either a local path (`/dev/sdb`, `./image.bin`) or @@ -67,29 +82,33 @@ clonetool sync --job NAME --source LOC --dest LOC [options] ``` # Same machine -clonetool sync --job disk1 --source /dev/sda --dest /dev/sdb +clonetool sync --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 +clonetool sync --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 +clonetool sync --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. | +| `--block-size` | `4M` | Block size (accepts `K`/`M`/`G` suffixes). | +| `--job` | — | Optional label shown in progress/log output. | | `--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. | +| `--sudo` | `auto` | Block-device privilege escalation: `auto` (on a permission error), `always`, or `never`. Remote elevation needs passwordless sudo. | +| `--deploy` | `true` | Copy this binary to remote hosts that lack a runnable `clonetool`. `--deploy=false` to disable. | | `--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). | +`clonetool version` prints the binary's `GOOS/GOARCH` (used internally for +the self-deploy check). + ## Caveats - Block-device size detection (`BLKGETSIZE64`) is Linux-only. @@ -98,6 +117,9 @@ Options: paths for typos before running. - SSH host keys are accepted on first connect (`StrictHostKeyChecking=accept-new`) and rejected if they later change, same as normal SSH behavior. +- Because every run re-hashes the whole destination, a re-sync costs a + full read of both sides even when little changed — the win is in the + bytes transferred, not the bytes read. - `agent` is an internal subcommand spawned automatically by `sync`; it's not meant to be run by hand, though it will work standalone for debugging. diff --git a/agent.go b/agent.go index ce797d2..c89f28d 100644 --- a/agent.go +++ b/agent.go @@ -1,8 +1,8 @@ package main import ( - "encoding/hex" "encoding/json" + "errors" "flag" "fmt" "io" @@ -63,17 +63,9 @@ func runControlAgent() error { case msgPrepare: handlePrepare(out, m) case msgConnectPush: - hashes, err := readHashTable(in) - if err != nil { - return err - } - runPushDriver(m, hashes, out) + runPushDriver(m, out) case msgConnectPull: - hashes, err := readHashTable(in) - if err != nil { - return err - } - runPullDriver(m, hashes, out) + runPullDriver(m, out) case msgClose: _ = out.WriteJSON(CtrlMsg{Type: msgBye}) return nil @@ -83,8 +75,9 @@ func runControlAgent() error { } } -func readHashTable(in *FrameReader) ([][32]byte, error) { - typ, payload, err := in.ReadFrame() +// readHashFrame reads one frameHashTable off fr and expands it. +func readHashFrame(fr *FrameReader) ([][32]byte, error) { + typ, payload, err := fr.ReadFrame() if err != nil { return nil, fmt.Errorf("read hash table: %w", err) } @@ -97,7 +90,7 @@ func readHashTable(in *FrameReader) ([][32]byte, error) { func handleStat(out *FrameWriter, m CtrlMsg) { info, err := statPath(m.Path) if err != nil { - _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error(), NeedPriv: isPermErr(err)}) return } _ = out.WriteJSON(CtrlMsg{Type: msgStatOK, Exists: info.Exists, IsDevice: info.IsDevice, Size: info.Size}) @@ -105,31 +98,51 @@ func handleStat(out *FrameWriter, m CtrlMsg) { func handlePrepare(out *FrameWriter, m CtrlMsg) { if err := prepareDest(m.Path, m.Size); err != nil { - _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error(), NeedPriv: isPermErr(err)}) return } _ = out.WriteJSON(CtrlMsg{Type: msgPrepareOK, Size: m.Size}) } +// isPermErr reports whether err (or anything it wraps, including a bare +// syscall errno from an ioctl) is a permission failure — the signal that +// retrying the agent under sudo might help. +func isPermErr(err error) bool { + return errors.Is(err, os.ErrPermission) +} + +// peerAgentCommand builds the *exec.Cmd for a one-shot helper (sink / +// source-stream) on the peer — locally or over ssh — wrapping it in sudo +// when req.Sudo is set (interactive sudo locally; "sudo -n" remotely, since +// a password prompt would corrupt the binary stream on stdout). +func peerAgentCommand(req CtrlMsg, tailArgs []string) *exec.Cmd { + if req.PeerLocal { + if req.Sudo { + return sudoLocalCommand(tailArgs) + } + return localAgentCommand(tailArgs) + } + peerArgs := append([]string{req.RemoteBin}, tailArgs...) + if req.Sudo { + peerArgs = append([]string{"sudo", "-n", "--"}, peerArgs...) + } + return sshCommand(req.SSHBin, req.SSHOpts, true, req.ConnectTimeoutSec, req.PeerUser, req.PeerHost, peerArgs) +} + // --------------------------------------------------------------------- // 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) { +func runPushDriver(req CtrlMsg, 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...)) - } + cmd := peerAgentCommand(req, tailArgs) stdin, err := cmd.StdinPipe() if err != nil { _ = out.WriteJSON(CtrlMsg{Type: msgPushFailed, Reason: err.Error()}) @@ -158,12 +171,22 @@ func runPushDriver(req CtrlMsg, hashes [][32]byte, out *FrameWriter) { return } - // Handshake succeeded: we're committed to push for this run. + // Handshake succeeded: we're committed to push for this run. The sink + // now sends the current per-block hashes of the destination it just + // read; the source loop compares against those to decide what to send. + hashes, err := readHashFrame(fr) + if err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: fmt.Sprintf("%v (remote stderr: %s)", err, stderrBuf.String())}) + return + } + srcFile, err := os.Open(req.Path) if err != nil { _ = cmd.Process.Kill() _ = cmd.Wait() - _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error()}) + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error(), NeedPriv: isPermErr(err)}) return } defer srcFile.Close() @@ -213,17 +236,14 @@ type ackEvent struct { // pumpPush runs the source-side read/hash/compare/send loop against fw // (the pipe to the remote sink) while concurrently draining ACK/ERR frames -// from fr, only reporting a block as done to the manager (over out) once -// its write has been confirmed. +// from fr, so it can wait for every sent block's write to be confirmed +// before declaring the push done. func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, fr *FrameReader, out *FrameWriter) error { const maxInFlight = 32 sem := make(chan struct{}, maxInFlight) var pendingMu sync.Mutex - pending := make(map[uint64][32]byte) - - batch := newResultBatcher(out) - defer batch.flush() + pending := make(map[uint64]bool) var copied, skipped int64 var lastProgress time.Time @@ -274,10 +294,10 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, 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) { + OnSend: func(idx uint64, _ [32]byte) { sem <- struct{}{} pendingMu.Lock() - pending[idx] = hash + pending[idx] = true pendingMu.Unlock() }, }) @@ -315,12 +335,11 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, ackCh = nil default: pendingMu.Lock() - h, ok := pending[ev.index] + _, 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() @@ -343,32 +362,26 @@ func pumpPush(req CtrlMsg, hashes [][32]byte, srcFile *os.File, fw *FrameWriter, return fmt.Errorf("sink closed the connection with %d block write confirmation(s) still outstanding", n) } - 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. +// pull driver: runs inside the DEST control agent. It reads and hashes the +// local destination itself, spawns ssh to the source host running the +// "source-stream" role, feeds it that hash table, then writes whatever it +// streams back straight 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) { +func runPullDriver(req CtrlMsg, 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...)) - } + cmd := peerAgentCommand(req, tailArgs) stdin, err := cmd.StdinPipe() if err != nil { _ = out.WriteJSON(CtrlMsg{Type: msgPullFailed, Reason: err.Error()}) @@ -397,7 +410,25 @@ func runPullDriver(req CtrlMsg, hashes [][32]byte, out *FrameWriter) { return } - // Handshake succeeded: send the hash table and commit to pull. + // Handshake succeeded. Open the destination, fingerprint its current + // content block by block, and hand that table to the source stream so it + // only sends back what differs. + 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(), NeedPriv: isPermErr(err)}) + return + } + defer dstFile.Close() + + hashes, err := hashFileBlocks(dstFile, req.Size, req.BlockSize) + if err != nil { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = out.WriteJSON(CtrlMsg{Type: msgError, Message: err.Error(), NeedPriv: isPermErr(err)}) + return + } if err := fw.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil { _ = cmd.Process.Kill() _ = cmd.Wait() @@ -405,30 +436,14 @@ func runPullDriver(req CtrlMsg, hashes [][32]byte, out *FrameWriter) { 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() @@ -459,9 +474,21 @@ func runSinkRole(path string, size, blockSize int64) error { } defer f.Close() + // Answer the handshake immediately so the push driver's short readiness + // timeout isn't spent hashing a large destination. if err := out.WriteFrame(frameReady, nil); err != nil { return err } + + hashes, err := hashFileBlocks(f, size, blockSize) + if err != nil { + fmt.Fprintf(os.Stderr, "sink: hash %s: %v\n", path, err) + return err + } + if err := out.WriteFrame(frameHashTable, flattenHashes(hashes)); err != nil { + return err + } + return runDestLoop(destLoopParams{File: f, BlockSize: blockSize, In: in, AckOut: out}) } @@ -515,43 +542,3 @@ func runSourceStreamRole(path string, size, blockSize int64) error { 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 index 8e89a55..61d1dfa 100644 --- a/control.go +++ b/control.go @@ -10,25 +10,39 @@ import ( "os/exec" ) +// errNeedPriv is wrapped into the error from a control-agent call that +// failed only because the agent lacked permission to open a device. +// bringUpController watches for it to decide whether to retry under sudo. +var errNeedPriv = errors.New("permission denied opening device") + // 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 + tag string + cmd *exec.Cmd + in io.WriteCloser + fw *FrameWriter + fr *FrameReader + sudo bool } -func startController(spec Spec, tag string, cfg *SyncConfig) (*Controller, error) { +func startController(spec Spec, tag string, cfg *SyncConfig, remoteBin string, sudo bool) (*Controller, error) { + agentArgs := []string{"agent", "--role", "control"} var cmd *exec.Cmd if spec.IsLocal() { - cmd = localAgentCommand([]string{"agent", "--role", "control"}) + if sudo { + cmd = sudoLocalCommand(agentArgs) + } else { + cmd = localAgentCommand(agentArgs) + } } else { - remoteArgs := []string{cfg.RemoteBin, "agent", "--role", "control"} + remoteArgs := append([]string{remoteBin}, agentArgs...) + if sudo { + remoteArgs = append([]string{"sudo", "-n", "--"}, remoteArgs...) + } cmd = sshCommand(cfg.SSHBin, cfg.SSHOpts, false, cfg.ConnectTimeoutSec, spec.User, spec.Host, remoteArgs) } stdin, err := cmd.StdinPipe() @@ -48,7 +62,7 @@ func startController(spec Spec, tag string, cfg *SyncConfig) (*Controller, error } go relayPrefixed(stderr, tag) - return &Controller{tag: tag, cmd: cmd, in: stdin, fw: NewFrameWriter(stdin), fr: NewFrameReader(stdout)}, nil + return &Controller{tag: tag, cmd: cmd, in: stdin, fw: NewFrameWriter(stdin), fr: NewFrameReader(stdout), sudo: sudo}, nil } func relayPrefixed(r io.Reader, tag string) { @@ -90,7 +104,7 @@ func (c *Controller) Stat(path string) (PathInfo, error) { 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) + return PathInfo{}, c.agentErr("stat", path, resp) default: return PathInfo{}, fmt.Errorf("%s: unexpected response %q to stat", c.tag, resp.Type) } @@ -105,17 +119,26 @@ func (c *Controller) Prepare(path string, size int64) error { case msgPrepareOK: return nil case msgError: - return fmt.Errorf("%s: prepare %s: %s", c.tag, path, resp.Message) + return c.agentErr("prepare", path, resp) default: return fmt.Errorf("%s: unexpected response %q to prepare", c.tag, resp.Type) } } +// agentErr turns an agent's error reply into an error, tagging it with +// errNeedPriv when the agent said the cause was a permission problem on a +// device (so the caller can retry the whole agent under sudo). +func (c *Controller) agentErr(op, path string, resp CtrlMsg) error { + if resp.NeedPriv { + return fmt.Errorf("%s: %s %s: %s: %w", c.tag, op, path, resp.Message, errNeedPriv) + } + return fmt.Errorf("%s: %s %s: %s", c.tag, op, path, resp.Message) +} + // transferCallbacks receives streaming updates while a connect_push or // connect_pull is in flight. type transferCallbacks struct { - onProgress func(CtrlMsg) - onBlockDone func(BlockResult) + onProgress func(CtrlMsg) } // ConnectPush asks the source control agent to try connecting straight out @@ -123,25 +146,22 @@ type transferCallbacks struct { // 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) { +func (c *Controller) ConnectPush(req CtrlMsg, cb transferCallbacks) (ok bool, reason string, err error) { req.Type = msgConnectPush - return c.connectAndPump(req, hashes, cb, msgPushOK, msgPushFailed) + return c.connectAndPump(req, 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) { +func (c *Controller) ConnectPull(req CtrlMsg, cb transferCallbacks) (ok bool, reason string, err error) { req.Type = msgConnectPull - return c.connectAndPump(req, hashes, cb, msgPullOK, msgPullFailed) + return c.connectAndPump(req, cb, msgPullOK, msgPullFailed) } -func (c *Controller) connectAndPump(req CtrlMsg, hashes [][32]byte, cb transferCallbacks, okType, failedType string) (bool, string, error) { +func (c *Controller) connectAndPump(req CtrlMsg, 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 { @@ -152,12 +172,6 @@ func (c *Controller) connectAndPump(req CtrlMsg, hashes [][32]byte, cb transferC 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: diff --git a/ctrlmsg.go b/ctrlmsg.go index 9b04f04..50b39c2 100644 --- a/ctrlmsg.go +++ b/ctrlmsg.go @@ -12,6 +12,9 @@ type CtrlMsg struct { Exists bool `json:"exists,omitempty"` IsDevice bool `json:"isDevice,omitempty"` Size int64 `json:"size,omitempty"` + // NeedPriv marks an error reply whose cause was a permission failure + // opening a device, so the manager can retry this agent under sudo. + NeedPriv bool `json:"needPriv,omitempty"` // connect_push (-> source agent) / connect_pull (-> dest agent) PeerHost string `json:"peerHost,omitempty"` @@ -28,44 +31,38 @@ type CtrlMsg struct { SSHOpts []string `json:"sshOpts,omitempty"` ConnectTimeoutSec int `json:"connectTimeoutSec,omitempty"` BlockSize int64 `json:"blockSize,omitempty"` + // Sudo tells a push/pull driver to run the peer helper it spawns + // (sink / source-stream) under "sudo -n" (remote) or "sudo" (local). + Sudo bool `json:"sudo,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"` + // progress + Copied int64 `json:"copied,omitempty"` + Skipped int64 `json:"skipped,omitempty"` + TotalBlocks int64 `json:"totalBlocks,omitempty"` + BytesCopied int64 `json:"bytesCopied,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" + 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" + msgLog = "log" + msgError = "error" + msgClose = "close" + msgBye = "bye" ) diff --git a/deploy.go b/deploy.go new file mode 100644 index 0000000..ae19651 --- /dev/null +++ b/deploy.go @@ -0,0 +1,109 @@ +package main + +import ( + "bytes" + "errors" + "fmt" + "os" + "os/exec" + "runtime" + "strings" +) + +// deployedRemoteBin is where clonetool copies itself on a remote host that +// doesn't already have a runnable one — a plain file under the user's home, +// no install, no root. +const deployedRemoteBin = ".clonetool/bin/clonetool" + +// resolveRemoteBin returns the path to a runnable clonetool on spec's host. +// It checks the configured --remote-bin first, then a copy deployed by an +// earlier run, and finally streams this binary over and verifies it runs. +func resolveRemoteBin(cfg *SyncConfig, spec Spec, tag string) (string, error) { + if _, code, err := runRemote(cfg, spec, cfg.RemoteBin, "version"); err != nil { + return "", fmt.Errorf("%s: cannot reach %s over ssh: %w", tag, spec.Host, err) + } else if code == 0 { + return cfg.RemoteBin, nil + } + + if !cfg.Deploy { + return "", fmt.Errorf("%s: %q is not runnable on %s and --deploy=false; "+ + "install clonetool there or pass --remote-bin", tag, cfg.RemoteBin, spec.Host) + } + + if _, code, err := runRemote(cfg, spec, deployedRemoteBin, "version"); err == nil && code == 0 { + return deployedRemoteBin, nil + } + + fmt.Fprintf(os.Stderr, "%s: no runnable clonetool on %s; copying this binary there ...\n", tag, spec.Host) + if err := deployBinary(cfg, spec); err != nil { + return "", fmt.Errorf("%s: copy clonetool to %s: %w", tag, spec.Host, err) + } + + out, code, err := runRemote(cfg, spec, deployedRemoteBin, "version") + if err != nil { + return "", fmt.Errorf("%s: verify clonetool on %s: %w", tag, spec.Host, err) + } + if code != 0 { + return "", fmt.Errorf( + "%s: copied clonetool to %s but it will not run there (%s). This binary is %s/%s; "+ + "build one for the remote's architecture (e.g. CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build) "+ + "and put it on PATH there or point --remote-bin at it", + tag, spec.Host, firstLine(out), runtime.GOOS, runtime.GOARCH) + } + fmt.Fprintf(os.Stderr, "%s: deployed to %s:~/%s (%s)\n", tag, spec.Host, deployedRemoteBin, strings.TrimSpace(out)) + return deployedRemoteBin, nil +} + +// runRemote runs `bin args...` on spec's host over ssh and returns its +// combined output and process exit code. A non-nil error means ssh itself +// could not run (connection/auth failure); a remote command that merely +// exits non-zero returns (output, code, nil). +func runRemote(cfg *SyncConfig, spec Spec, bin string, args ...string) (string, int, error) { + remoteArgs := append([]string{bin}, args...) + cmd := sshCommand(cfg.SSHBin, cfg.SSHOpts, false, cfg.ConnectTimeoutSec, spec.User, spec.Host, remoteArgs) + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + err := cmd.Run() + if err == nil { + return buf.String(), 0, nil + } + var ee *exec.ExitError + if errors.As(err, &ee) { + return buf.String(), ee.ExitCode(), nil + } + return buf.String(), -1, err +} + +// deployBinary streams this executable to ~/.clonetool/bin/clonetool on +// spec's host, writing to a temp name and renaming into place so a +// concurrent run never sees a half-written file. +func deployBinary(cfg *SyncConfig, spec Spec) error { + f, err := os.Open(selfExe()) + if err != nil { + return err + } + defer f.Close() + + const script = `set -e; d="$HOME/.clonetool/bin"; mkdir -p "$d"; ` + + `cat > "$d/clonetool.new"; chmod 755 "$d/clonetool.new"; mv "$d/clonetool.new" "$d/clonetool"` + cmd := sshCommand(cfg.SSHBin, cfg.SSHOpts, false, cfg.ConnectTimeoutSec, spec.User, spec.Host, []string{"sh", "-c", script}) + cmd.Stdin = f + var errb bytes.Buffer + cmd.Stderr = &errb + if err := cmd.Run(); err != nil { + if msg := strings.TrimSpace(errb.String()); msg != "" { + return fmt.Errorf("%w: %s", err, firstLine(msg)) + } + return err + } + return nil +} + +func firstLine(s string) string { + s = strings.TrimSpace(s) + if i := strings.IndexByte(s, '\n'); i >= 0 { + return s[:i] + } + return s +} diff --git a/jobstate.go b/jobstate.go deleted file mode 100644 index 592f446..0000000 --- a/jobstate.go +++ /dev/null @@ -1,204 +0,0 @@ -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 index 9482f79..446f1de 100644 --- a/main.go +++ b/main.go @@ -4,6 +4,7 @@ import ( "flag" "fmt" "os" + "runtime" ) func main() { @@ -18,6 +19,9 @@ func main() { err = cmdSync(os.Args[2:]) case "agent": err = cmdAgent(os.Args[2:]) + case "version", "--version": + fmt.Printf("clonetool %s/%s\n", runtime.GOOS, runtime.GOARCH) + return case "-h", "--help", "help": usage() return @@ -35,7 +39,8 @@ func usage() { fmt.Fprint(os.Stderr, `clonetool - block-level file/device sync Usage: - clonetool sync --job NAME --source LOC --dest LOC [options] + clonetool sync --source LOC --dest LOC [options] + clonetool version 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. @@ -43,21 +48,23 @@ 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. +clonetool keeps no state between runs: every sync re-reads and re-hashes both +the source and the destination and transfers only the blocks that differ. + 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) + --job NAME optional label shown in progress/log output --yes don't prompt before shrinking an existing destination file - --force rebind this job to a new source/dest, discarding hash history + --sudo MODE device-access privilege escalation: auto (escalate on a + permission error, default), always, or never + --deploy copy this binary to remote hosts that lack it + (default true; --deploy=false to disable) --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. `) } @@ -71,13 +78,13 @@ func (s *stringSlice) Set(v string) error { func cmdSync(args []string) error { fs := flag.NewFlagSet("sync", flag.ContinueOnError) - job := fs.String("job", "", "job name (required)") + job := fs.String("job", "", "optional label for progress/log output") 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") + sudoMode := fs.String("sudo", "auto", "device-access privilege escalation: auto|always|never") + deploy := fs.Bool("deploy", true, "copy this binary to remote hosts that lack it") 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") @@ -88,19 +95,24 @@ func cmdSync(args []string) error { return err } - if *job == "" || *source == "" || *dest == "" { + if *source == "" || *dest == "" { fs.Usage() - return fmt.Errorf("--job, --source and --dest are required") + return fmt.Errorf("--source and --dest are required") } blockSize, err := parseSize(*blockSizeStr) if err != nil { return fmt.Errorf("--block-size: %w", err) } + switch *sudoMode { + case "auto", "always", "never": + default: + return fmt.Errorf("--sudo: want auto|always|never, got %q", *sudoMode) + } 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, + Job: *job, Source: *source, Dest: *dest, BlockSize: blockSize, + Yes: *yes, Sudo: *sudoMode, Deploy: *deploy, ConnectTimeoutSec: *connectTimeout, + SSHBin: *sshBin, SSHOpts: sshOpts, RemoteBin: *remoteBin, ManagerHost: *managerHost, }) } diff --git a/manager.go b/manager.go index 48b7c03..ebd9bb0 100644 --- a/manager.go +++ b/manager.go @@ -2,11 +2,10 @@ package main import ( "bufio" - "encoding/hex" + "errors" "fmt" "os" "strings" - "time" ) const defaultBlockSize = 4 * 1024 * 1024 @@ -20,9 +19,9 @@ type SyncConfig struct { Source string Dest string BlockSize int64 - StateDir string Yes bool - Force bool + Sudo string // auto | always | never + Deploy bool // copy this binary to remote hosts that lack it ConnectTimeoutSec int SSHBin string SSHOpts []string @@ -43,30 +42,37 @@ func runSync(cfg SyncConfig) error { return err } - srcCtrl, err := startController(srcSpec, "source", &cfg) + // Make sure each remote endpoint has a runnable clonetool, copying this + // binary over if not (unless --deploy=false). A host that appears on + // both sides is only probed once. + srcRemoteBin, dstRemoteBin := cfg.RemoteBin, cfg.RemoteBin + if !srcSpec.IsLocal() { + if srcRemoteBin, err = resolveRemoteBin(&cfg, srcSpec, "source"); err != nil { + return err + } + } + if !dstSpec.IsLocal() { + if !srcSpec.IsLocal() && sameHost(srcSpec, dstSpec) { + dstRemoteBin = srcRemoteBin + } else if dstRemoteBin, err = resolveRemoteBin(&cfg, dstSpec, "dest"); err != nil { + return err + } + } + + srcCtrl, srcInfo, srcSudo, err := bringUpController(srcSpec, "source", &cfg, srcRemoteBin, srcSpec.Path) if err != nil { - return fmt.Errorf("start source control agent: %w", err) + return fmt.Errorf("source: %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) + dstCtrl, dstInfo, dstSudo, err := bringUpController(dstSpec, "dest", &cfg, dstRemoteBin, dstSpec.Path) if err != nil { - return fmt.Errorf("stat destination: %w", err) + return fmt.Errorf("dest: %w", err) } + defer dstCtrl.Close() targetSize, err := computeTargetSize(srcSpec, srcInfo, dstSpec, dstInfo, cfg.Yes) if err != nil { @@ -78,41 +84,8 @@ func runSync(cfg SyncConfig) error { } 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) - }, - } + cb := transferCallbacks{onProgress: func(m CtrlMsg) { printProgress(m) }} bothLocal := srcSpec.IsLocal() && dstSpec.IsLocal() srcHost, srcUser := resolveConnectHost(srcSpec, &cfg) @@ -121,14 +94,15 @@ func runSync(cfg SyncConfig) error { 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, + RemoteBin: dstRemoteBin, SSHBin: cfg.SSHBin, SSHOpts: cfg.SSHOpts, ConnectTimeoutSec: cfg.ConnectTimeoutSec, + Sudo: dstSudo, } 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) + ok, reason, err := srcCtrl.ConnectPush(pushReq, cb) if err != nil { return err } @@ -137,9 +111,10 @@ func runSync(cfg SyncConfig) error { 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, + RemoteBin: srcRemoteBin, SSHBin: cfg.SSHBin, SSHOpts: cfg.SSHOpts, ConnectTimeoutSec: cfg.ConnectTimeoutSec, + Sudo: srcSudo, } - ok2, reason2, err2 := dstCtrl.ConnectPull(pullReq, state.Hashes, cb) + ok2, reason2, err2 := dstCtrl.ConnectPull(pullReq, cb) if err2 != nil { return err2 } @@ -152,12 +127,60 @@ func runSync(cfg SyncConfig) error { } 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) + label := cfg.Job + if label == "" { + label = fmt.Sprintf("%s -> %s", srcSpec, dstSpec) + } + fmt.Fprintf(os.Stderr, "done: %s, %d blocks\n", label, blockCount) return nil } +// bringUpController starts a control agent for spec and does its initial +// stat. With --sudo=always the agent is elevated from the start; with +// --sudo=auto a permission error on the stat triggers one transparent +// restart under sudo. The returned bool reports whether the agent (and any +// peer helper it later spawns for this side) is running elevated. +func bringUpController(spec Spec, tag string, cfg *SyncConfig, remoteBin, probePath string) (*Controller, PathInfo, bool, error) { + sudo := cfg.Sudo == "always" + c, err := startController(spec, tag, cfg, remoteBin, sudo) + if err != nil { + return nil, PathInfo{}, sudo, err + } + info, err := c.Stat(probePath) + if err == nil { + return c, info, sudo, nil + } + if errors.Is(err, errNeedPriv) && cfg.Sudo == "auto" && !sudo { + fmt.Fprintf(os.Stderr, "%s: permission denied on %s; retrying via sudo ...\n", tag, probePath) + c.Close() + sudo = true + if c, err = startController(spec, tag, cfg, remoteBin, true); err != nil { + return nil, PathInfo{}, sudo, err + } + if info, err = c.Stat(probePath); err == nil { + return c, info, sudo, nil + } + } + c.Close() + if sudo { + return nil, PathInfo{}, sudo, fmt.Errorf( + "%w; if this is a sudo failure, configure passwordless sudo for clonetool on %s or run the manager as root", + err, hostLabel(spec)) + } + return nil, PathInfo{}, sudo, err +} + +func sameHost(a, b Spec) bool { + return strings.EqualFold(a.Host, b.Host) && a.User == b.User +} + +func hostLabel(spec Spec) string { + if spec.IsLocal() { + return "this machine" + } + return spec.Host +} + func resolveConnectHost(spec Spec, cfg *SyncConfig) (host, user string) { if !spec.IsLocal() { return spec.Host, spec.User diff --git a/syncside.go b/syncside.go index d3a64a5..d38efc5 100644 --- a/syncside.go +++ b/syncside.go @@ -27,6 +27,24 @@ func readBlockAt(f *os.File, buf []byte, offset, totalSize int64) (int, error) { return n, nil } +// hashFileBlocks reads f in blockSize-byte blocks up to size and returns the +// SHA-256 of each. This is how a destination fingerprints its *current* +// content at the start of every sync — clonetool keeps no hash state of its +// own between runs. +func hashFileBlocks(f *os.File, size, blockSize int64) ([][32]byte, error) { + blockCount := (size + blockSize - 1) / blockSize + out := make([][32]byte, blockCount) + buf := make([]byte, blockSize) + for i := int64(0); i < blockCount; i++ { + n, err := readBlockAt(f, buf, i*blockSize, size) + if err != nil { + return nil, fmt.Errorf("hash %s at block %d: %w", f.Name(), i, err) + } + out[i] = sha256.Sum256(buf[:n]) + } + return out, 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 @@ -79,7 +97,6 @@ type destLoopParams struct { 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. @@ -118,9 +135,6 @@ func runDestLoop(p destLoopParams) error { return err } } - if p.OnWritten != nil { - p.OnWritten(index, hash) - } case frameCtrlJSON: if p.OnCtrlMsg != nil { var m CtrlMsg diff --git a/transport.go b/transport.go index 6c00bff..e873dae 100644 --- a/transport.go +++ b/transport.go @@ -46,12 +46,24 @@ func sshCommand(sshBin string, extraOpts []string, batchMode bool, connectTimeou 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 { +// selfExe is the path to the running clonetool binary — used to re-exec it +// as a local agent and to stream it to a remote host for deployment. +func selfExe() string { self, err := os.Executable() if err != nil { - self = os.Args[0] + return os.Args[0] } - return exec.Command(self, remoteArgs...) + return self +} + +// localAgentCommand re-execs this same binary as an agent, for a spec with +// no host part. +func localAgentCommand(agentArgs []string) *exec.Cmd { + return exec.Command(selfExe(), agentArgs...) +} + +// sudoLocalCommand is localAgentCommand wrapped in an interactive sudo: the +// manager still owns the terminal, so a local password prompt works. +func sudoLocalCommand(agentArgs []string) *exec.Cmd { + return exec.Command("sudo", append([]string{"--", selfExe()}, agentArgs...)...) }