diff --git a/README.md b/README.md index 4f3ea15..f8dff34 100644 --- a/README.md +++ b/README.md @@ -11,16 +11,19 @@ really walks, filters, diffs, transfers, and reconstructs files, applying requested attributes along the way. See [End-to-End Sync Pipeline](#end-to-end-sync-pipeline) below for exactly how the pieces connect and, just as importantly, what's still explicitly -out of scope (compression, partial/append transfers, batch mode, full -`--delete`, and device/special files) - this is real, working sync, not -yet full feature parity. Hard links *are* now preserved, opt-in via -`-H`/`--hard-links` exactly like real rsync's own flag (see +out of scope (partial/append transfers, batch mode, full `--delete`, and +device/special files) - this is real, working sync, not yet full feature +parity. Hard links *are* now preserved, opt-in via `-H`/`--hard-links` +exactly like real rsync's own flag (see [File Attribute Preservation](#file-attribute-preservation) below), and `--dry-run`/`-n` is a genuine trial run - full planning, zero filesystem changes - with real `--itemize-changes`/`-i` output matching rsync's own format (see [Dry-Run Mode](#dry-run-mode) below). `--progress` and `--stats` are now implemented too, both matching real rsync's own output formats (see [Progress and Stats](#progress-and-stats) below). +`--compress`/`-z` now genuinely compresses a file's literal delta data +with zlib, including real rsync's own `--compress-level` and +`--skip-compress` (see [Compression](#compression) below). `grsync --daemon` also now speaks a real subset of the rsync daemon protocol - `rsyncd.conf` parsing, the `@RSYNCD` greeting/handshake, module @@ -671,6 +674,156 @@ higher-priority branch. Locked in by inert for this direction (`Sender` never even looks at `ReceiverOptions`), and the upload itself still completes correctly. +## Compression + +`--compress`/`-z` compresses a file's literal delta data with zlib +before it crosses the wire; `--compress-level` controls how hard, and +`--skip-compress` excludes already-compressed file types. All three +match real rsync's own semantics, verified against upstream's actual +source (`token.c`, `RsyncProject/rsync`) and `rsync.1`'s own documented +wording rather than assumed. + +### What gets compressed, and what doesn't + +Only a delta's literal data - the bytes a `DataOp` carries because they +didn't match anything in the receiver's signature - is ever compressed. +`CopyOp`s (plain block-index references, not data) and every +`FrameSignature` (checksums) are never touched, matching the ticket's +own scope exactly. + +Rather than compressing each `DataOp` independently, `Sender` (via +`toWireDeltaOps`, `internal/pipeline/messages.go`) concatenates *all* of +one file's literal data into a single buffer and zlib-compresses that as +one unit per `FrameDelta` message, with a single `Compressed bool` +marker on the message itself; each `DataOp`'s own wire form then carries +only how many of the decompressed stream's bytes are its +(`wireDeltaOp.Length`), so the receiver can re-slice it back apart after +one decompression. This amortizes zlib's fixed ~8-byte header/trailer +overhead across a whole file instead of paying it again for every +separate literal run a scattered-changes file can produce - closer to +real rsync's own `zlibx` compression choice (one persistent per-file +deflate stream with matched data excluded from it) than compressing +op-by-op would have been. If the compressed result isn't actually +smaller than the raw literal data - realistic for a file whose total +changed content is tiny, exactly the case delta transfer exists for, or +for already-incompressible data that slipped past `--skip-compress` - +the file is sent uncompressed instead; `Compressed` is a real, checked +outcome, not just a request. + +Compression is entirely a **sending-side** decision. `Receiver` takes no +compression-related options of its own at all: it simply decompresses +whatever each `deltaMessage.Compressed` marker says, on every transport, +which is possible because grsync only ever pushes (see +[Status](#status)) - `Sender` always runs on the local, requesting +process, never remotely, for every path currently wired into the CLI. + +### `--compress-level` + +Verified against real rsync's own source (`token.c`'s +`init_compression_level`) and `rsync.1`'s own documented wording: for +zlib compression, valid levels are **1 (fastest) to 9 (smallest), with 6 +as the default**. `--compress-level=0` explicitly turns compression off +- overriding a `-z` given alongside it - and `--compress-level=-1` means +"use the default." Giving `--compress-level` alone, without `-z`, +implies compression (unless the resulting level is 0), matching real +rsync's own documented "the `--compress` option is implied" rule. An +out-of-range value is silently clamped into `[1, 9]`, matching +`rsync.1`'s own "too-large or too-small value" wording. All of this is +`ClampCompressLevel` (`internal/pipeline/compress.go`) and +`effectiveCompressOptions` (`internal/cli/sync.go`). + +### `--skip-compress` + +Overrides the built-in list of already-compressed file suffixes +(`gz`, `zip`, `jpg`, `mp3`, `mp4`, and 91 others) that are sent +uncompressed even with `--compress` on, since running zlib over an +already-compressed format wastes CPU for no size benefit. The full +default list is real rsync's own, copied verbatim from `rsync.1`'s own +documented default (`DefaultSkipCompressSuffixes`, +`internal/pipeline/compress.go`) rather than invented. An explicit +`--skip-compress=""` is a meaningful override in its own right ("skip +nothing"), matching real rsync's own documented meaning for it - grsync +tells that apart from "the flag was never given at all" via +`cmd.Flags().Changed`, not `opts.skipCompress`'s zero value, since an +empty string is both. + +Matching is a plain, case-insensitive suffix list +(`--skip-compress=gz/jpg/mp3`); real rsync's own `--skip-compress` +grammar additionally supports bracketed character classes inside a +suffix (e.g. `mp[34]`), which grsync's version does not - a deliberate, +disclosed scope reduction, since plain suffixes cover the default list +and the overwhelming majority of real-world uses. + +**Worth disclosing**: real rsync's own current documentation (as of this +writing) admits `--skip-compress` "has no effect" in its own latest +implementation, because none of its currently-supported compression +algorithms allow changing level mid-stream - its per-file persistent +deflate context, once opened, keeps compressing everything at the same +level regardless of what the suffix list says. grsync's frame-per-file +design has no such persistent stream to be stuck with: `toWireDeltaOps` +makes a fresh, genuine "compress this file's literal data or don't" call +for every file, so `--skip-compress` actually works here - a real +improvement made possible by the architectural difference, not a silent +divergence from upstream's documented behavior. + +### Interaction with `--stats` and `--dry-run` + +`--stats`' "Total bytes sent"/"Total bytes received" (see +[Progress and Stats](#progress-and-stats)) already measure genuine wire +traffic via `countingReadWriter`, which wraps the connection itself - so +compressed bytes are reflected automatically, no changes needed for this +ticket. Since `Stats` is computed on the *receiving* side, it's +specifically **"Total bytes received"** that shrinks with compression +for an upload (`Sender` on the far end sends compressed data, this side +receives it) - "Total bytes sent" reflects this side's own small +signature/ack traffic back to the sender, which compression doesn't +touch. `Total file size` is unaffected either way, since it describes +the files themselves, not what crossed the wire. + +`--dry-run` and `--compress` compose cleanly: the full signature/delta +exchange - including compressing the delta's literal data - still runs +during a dry run exactly as it would for a real sync (see +[Dry-Run Mode](#dry-run-mode)'s own explanation of why), so itemize +output stays accurate; only the final disk write is skipped, and +compression has nothing to do with that. + +### Across transports + +- **Local**: `Sender` runs in-process with the `CompressOptions` + `effectiveCompressOptions` computed from the CLI flags - no different + from any other in-process call. +- **SSH**: unlike `--dry-run`/`--itemize-changes`/`--verbose`/ + `--progress`/`--stats` (see [Dry-Run Mode](#dry-run-mode)'s own + "Across transports" section), `--compress` needs **no remote argv + change at all**: `Sender` runs locally for this transport too, and the + remote `--server` process's `Receiver` just reacts to each + `deltaMessage`'s own `Compressed` marker, exactly like every other + transport. Verified over a real SSH connection to `127.0.0.1` by + `TestSSHLocalhost_CompressDoesNotBreakTheTransfer` (skipped gracefully + without a local `sshd`). +- **`rsync://` daemon, upload (`DirectionPut`)**: `Sender` runs on the + *client* side for this direction (see + [rsync Daemon Mode](#rsync-daemon-mode)), exactly where `--compress`'s + decision belongs - no daemon-protocol extension needed, the same way + SSH needs none. Verified over a real TCP connection by + `TestDaemon_RealTCP_PutWithCompressUploadsCorrectly`. +- **`rsync://` daemon, download (`DirectionGet`)**: the daemon's own + `Sender` call for a module download has no CLI wiring at all yet - see + [Status](#status)'s "push only" scope boundary, already established + before this ticket - so it always runs with compression disabled + (`pipeline.CompressOptions{}`), consistent with that existing + boundary, not a new gap introduced here. + +### Hard links + +A hard-link group's secondary members never go through the +signature/delta exchange at all - `Sender` and `Receiver` both skip them +outright, recreating the link directly instead (see +[File Attribute Preservation](#file-attribute-preservation)) - so +`--compress` is moot for them by construction, with nothing to compress +in the first place. `TestSenderReceiver_CompressWorksWithHardLinks` +confirms that skip still holds correctly with compression enabled. + ## rsync Daemon Mode `internal/daemon` implements grsync's `--daemon` server mode: a second way diff --git a/internal/cli/root.go b/internal/cli/root.go index cb223ee..493c3ce 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -41,29 +41,31 @@ type FilterRule struct { // in one struct (rather than loose variables) makes it straightforward to // pass a single value into internal/sync once that package exists. type options struct { - archive bool - verbose bool - compress bool - recursive bool - dirs bool - dryRun bool - delete bool - progress bool - perms bool - times bool - owner bool - group bool - links bool - hardLinks bool - itemize bool - stats bool - filterRules []FilterRule - rsh string - server bool - daemon bool - config string - port int - passwordFile string + archive bool + verbose bool + compress bool + compressLevel int + skipCompress string + recursive bool + dirs bool + dryRun bool + delete bool + progress bool + perms bool + times bool + owner bool + group bool + links bool + hardLinks bool + itemize bool + stats bool + filterRules []FilterRule + rsh string + server bool + daemon bool + config string + port int + passwordFile string } // filterRuleFlag implements pflag.Value. Each of --exclude/--include/ @@ -106,7 +108,7 @@ func NewRootCmd() *cobra.Command { Short: "grsync synchronizes files between one or more sources and a destination", Long: "grsync is an rsync-inspired file synchronization tool.\n" + "Local-to-local and local-to-remote (SSH) syncs are supported, including --dry-run, " + - "--itemize-changes, --progress, and --stats; compression and full --delete are not yet.", + "--itemize-changes, --progress, --stats, and --compress; full --delete is not yet.", // --server takes exactly one positional arg (the destination path) // rather than the normal ... shape: it is how // a remote-invoked grsync (e.g. `ssh host grsync --server /dest`) @@ -141,7 +143,18 @@ func NewRootCmd() *cobra.Command { flags.BoolVarP(&opts.verbose, "verbose", "v", false, "mention each updated item's path (superseded by --itemize-changes when both are given, "+ "matching real rsync's own -v/-i relationship)") - flags.BoolVarP(&opts.compress, "compress", "z", false, "compress file data during transfer") + flags.BoolVarP(&opts.compress, "compress", "z", false, + "compress a file's literal delta data with zlib before sending it (checksums/signatures are never "+ + "compressed); see the README's Compression section") + flags.IntVar(&opts.compressLevel, "compress-level", 0, + "explicitly set the zlib compression level (1-9, default 6) instead of --compress's implicit "+ + "default; implies --compress even without -z, unless set to 0 (\"off\", which disables "+ + "compression even if -z was also given) - matches real rsync's own --compress-level/--zl "+ + "range, default, and off-semantics, verified against upstream source") + flags.StringVar(&opts.skipCompress, "skip-compress", "", + "override the default list of already-compressed file suffixes (slash-separated, e.g. gz/jpg/mp3) "+ + "that --compress/-z sends uncompressed; an empty string means \"skip nothing\" - matches real "+ + "rsync's own --skip-compress (see the README's Compression section for the full built-in default list)") flags.BoolVarP(&opts.recursive, "recursive", "r", false, "recurse into directories") flags.BoolVarP(&opts.dirs, "dirs", "d", false, "include directories themselves without recursing into their contents (implied by --recursive)") flags.BoolVarP(&opts.dryRun, "dry-run", "n", false, "perform a trial run: full planning (file list, filters, deltas) with no filesystem changes") diff --git a/internal/cli/rsync_url.go b/internal/cli/rsync_url.go index f7cfb43..61e1dec 100644 --- a/internal/cli/rsync_url.go +++ b/internal/cli/rsync_url.go @@ -44,7 +44,14 @@ const dialDaemonTimeout = 10 * time.Second // Itemize/Verbose/Progress/Stats are all deliberately left unset - see // runSync's own one-time note about why none of daemon-PUT's reporting // output is available, printed before this function is ever called. -func syncToRsyncDaemon(src string, u daemon.URL, password daemon.PasswordFunc, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool, dryRun bool) error { +// +// copts, in contrast, is fully honored here: DirectionPut runs Sender on +// this (the client) side (see DialModule's own doc comment), exactly +// where --compress/-z's decision actually belongs - no wire extension +// like dryRunToken is needed for it at all, since the server's Receiver +// only ever reacts to what each deltaMessage's own Compressed marker +// says. +func syncToRsyncDaemon(src string, u daemon.URL, password daemon.PasswordFunc, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool, dryRun bool, copts pipeline.CompressOptions) error { port := u.Port if port == 0 { port = daemon.DefaultPort @@ -59,5 +66,5 @@ func syncToRsyncDaemon(src string, u daemon.URL, password daemon.PasswordFunc, w user := resolveUser(u.User) ropts := pipeline.ReceiverOptions{DryRun: dryRun} - return daemon.DialClient(nc, u.Module, user, password, daemon.DirectionPut, src, rules, walkOpts, sync.AttrOptions{HardLinks: hardLinks}, ropts) + return daemon.DialClient(nc, u.Module, user, password, daemon.DirectionPut, src, rules, walkOpts, sync.AttrOptions{HardLinks: hardLinks}, ropts, copts) } diff --git a/internal/cli/sync.go b/internal/cli/sync.go index 9f4482e..6c03e34 100644 --- a/internal/cli/sync.go +++ b/internal/cli/sync.go @@ -63,6 +63,43 @@ func effectiveReceiverOptions(opts *options, output io.Writer) pipeline.Receiver } } +// effectiveCompressOptions computes pipeline.CompressOptions from opts, +// mirroring real rsync's own --compress-level implication rule ("The +// --compress option is implied as long as the level chosen is not a +// 'don't compress' level" - rsync.1): compression is enabled whenever +// either -z or --compress-level was given at all, at whatever level +// --compress-level requested (clamped) or the real-rsync-verified +// default of 6 if only -z was given, UNLESS that level clamps to 0 +// ("off"), which disables compression outright even if -z was also +// given - matching real rsync's own documented "--zl=0 turns compression +// off" behavior exactly, including its override of -z. +// +// cmd.Flags().Changed, not opts.compressLevel's zero value, is what +// distinguishes "--compress-level was never given" from "--compress-level=0 +// was given explicitly" - those two cases mean different things (default +// level 6 vs. explicitly off) and pflag's own IntVar can't tell them +// apart by value alone, since 0 is also its unset zero value. +func effectiveCompressOptions(cmd *cobra.Command, opts *options) pipeline.CompressOptions { + levelGiven := cmd.Flags().Changed("compress-level") + if !opts.compress && !levelGiven { + return pipeline.CompressOptions{} + } + + level := pipeline.DefaultCompressLevel + if levelGiven { + level = pipeline.ClampCompressLevel(opts.compressLevel) + } + if level == 0 { + return pipeline.CompressOptions{} + } + + suffixes := pipeline.DefaultSkipCompressSuffixes + if cmd.Flags().Changed("skip-compress") { + suffixes = pipeline.ParseSkipCompressList(opts.skipCompress) + } + return pipeline.CompressOptions{Enabled: true, Level: level, SkipSuffixes: suffixes} +} + // toSyncRawRules converts the CLI's FilterRule list to sync.RawRule. // FilterRuleType's string values were chosen to exactly match // sync.RuleKind's ("include", "exclude", "filter", "exclude-from", @@ -102,6 +139,7 @@ func runSync(cmd *cobra.Command, sources []string, destination string, opts *opt walkOpts := effectiveWalkOptions(opts) attrOpts := effectiveAttrOptions(opts) + copts := effectiveCompressOptions(cmd, opts) rules, err := sync.CompileRules(toSyncRawRules(opts.filterRules)) if err != nil { return fmt.Errorf("compiling filter rules: %w", err) @@ -161,15 +199,15 @@ func runSync(cmd *cobra.Command, sources []string, destination string, opts *opt for _, src := range sources { switch { case isRsyncDaemon: - if err := syncToRsyncDaemon(src, rsyncURL, password, walkOpts, rules, attrOpts.HardLinks, opts.dryRun); err != nil { + if err := syncToRsyncDaemon(src, rsyncURL, password, walkOpts, rules, attrOpts.HardLinks, opts.dryRun, copts); err != nil { return fmt.Errorf("syncing %q to %q: %w", src, destination, err) } case isRemote: - if err := syncToRemote(opts.rsh, src, remote, walkOpts, rules, attrOpts.HardLinks, ropts); err != nil { + if err := syncToRemote(opts.rsh, src, remote, walkOpts, rules, attrOpts.HardLinks, ropts, copts); err != nil { return fmt.Errorf("syncing %q to %q: %w", src, destination, err) } default: - if err := syncLocal(src, destination, walkOpts, rules, attrOpts, ropts); err != nil { + if err := syncLocal(src, destination, walkOpts, rules, attrOpts, ropts, copts); err != nil { return fmt.Errorf("syncing %q to %q: %w", src, destination, err) } } @@ -188,7 +226,7 @@ func runSync(cmd *cobra.Command, sources []string, destination string, opts *opt // way, the exact same pipeline.Sender/pipeline.Receiver functions that // carry out a remote sync are what a local sync exercises too, instead of // a second, independently-trusted implementation of the same logic. -func syncLocal(src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions) error { +func syncLocal(src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions, copts pipeline.CompressOptions) error { senderReadsFromReceiver, receiverWritesToSender := io.Pipe() receiverReadsFromSender, senderWritesToReceiver := io.Pipe() @@ -196,7 +234,7 @@ func syncLocal(src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, a receiver := pipeReadWriter{Reader: receiverReadsFromSender, Writer: receiverWritesToSender} senderErrCh := make(chan error, 1) - go func() { senderErrCh <- pipeline.Sender(sender, src, walkOpts, rules, attrOpts.HardLinks) }() + go func() { senderErrCh <- pipeline.Sender(sender, src, walkOpts, rules, attrOpts.HardLinks, copts) }() receiverErr := pipeline.Receiver(receiver, dest, attrOpts, ropts) senderErr := <-senderErrCh @@ -221,7 +259,14 @@ func syncLocal(src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, a // here at all. This is the same mechanism SC-11 established for // DryRun/Itemize/Verbose; Progress/Stats just reuse it rather than // inventing a second one. -func syncToRemote(rsh, src string, remote transport.RemotePath, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool, ropts pipeline.ReceiverOptions) error { +// +// copts needs none of that: --compress/-z is a Sender-side decision (see +// pipeline.CompressOptions' own doc comment), and Sender runs right here, +// locally, for this transport - there is nothing for the remote +// --server process to be told via argv at all. The remote Receiver just +// decompresses whatever each deltaMessage's own Compressed marker says, +// exactly like every other transport. +func syncToRemote(rsh, src string, remote transport.RemotePath, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool, ropts pipeline.ReceiverOptions, copts pipeline.CompressOptions) error { remoteArgs := []string{"grsync", "--server"} if ropts.DryRun { remoteArgs = append(remoteArgs, "--dry-run") @@ -250,7 +295,7 @@ func syncToRemote(rsh, src string, remote transport.RemotePath, walkOpts sync.Wa return fmt.Errorf("handshake with %s failed: %w", remote.Host, err) } - sendErr := pipeline.Sender(session, src, walkOpts, rules, hardLinks) + sendErr := pipeline.Sender(session, src, walkOpts, rules, hardLinks, copts) closeErr := session.Close() if sendErr != nil { diff --git a/internal/cli/sync_test.go b/internal/cli/sync_test.go index 9b204dc..ee59609 100644 --- a/internal/cli/sync_test.go +++ b/internal/cli/sync_test.go @@ -5,13 +5,128 @@ import ( "io/fs" "os" "path/filepath" + "regexp" "runtime" + "strconv" "strings" "testing" + "github.com/spf13/cobra" + + "github.com/syntaxroot-cc/grsync/internal/pipeline" "github.com/syntaxroot-cc/grsync/internal/sync" ) +// buildCompressTestCmd registers only the three --compress-related flags +// (not NewRootCmd's full set) on a bare *cobra.Command and parses args +// against them, giving effectiveCompressOptions_test.go's table test +// direct control over cmd.Flags().Changed - the thing that actually +// distinguishes "--compress-level was never given" from "--compress-level=0 +// was given explicitly," which opts.compressLevel's own zero value can't +// do alone (see effectiveCompressOptions' own doc comment). +func buildCompressTestCmd(t *testing.T, args []string) (*cobra.Command, *options) { + t.Helper() + opts := &options{} + cmd := &cobra.Command{} + flags := cmd.Flags() + flags.BoolVarP(&opts.compress, "compress", "z", false, "") + flags.IntVar(&opts.compressLevel, "compress-level", 0, "") + flags.StringVar(&opts.skipCompress, "skip-compress", "", "") + if err := flags.Parse(args); err != nil { + t.Fatalf("Parse(%v) returned error: %v", args, err) + } + return cmd, opts +} + +func TestEffectiveCompressOptions(t *testing.T) { + tests := []struct { + name string + args []string + want pipeline.CompressOptions + }{ + { + name: "nothing given", + args: nil, + want: pipeline.CompressOptions{}, + }, + { + name: "-z alone uses the real-rsync-verified default level and default suffix list", + args: []string{"-z"}, + want: pipeline.CompressOptions{Enabled: true, Level: pipeline.DefaultCompressLevel, SkipSuffixes: pipeline.DefaultSkipCompressSuffixes}, + }, + { + name: "--compress-level alone implies --compress, matching real rsync", + args: []string{"--compress-level=9"}, + want: pipeline.CompressOptions{Enabled: true, Level: 9, SkipSuffixes: pipeline.DefaultSkipCompressSuffixes}, + }, + { + name: "--compress-level=0 alone does not enable compression", + args: []string{"--compress-level=0"}, + want: pipeline.CompressOptions{}, + }, + { + name: "--compress-level=0 overrides an explicit -z, matching real rsync's own documented behavior", + args: []string{"-z", "--compress-level=0"}, + want: pipeline.CompressOptions{}, + }, + { + name: "--compress-level=-1 means \"use the default\"", + args: []string{"-z", "--compress-level=-1"}, + want: pipeline.CompressOptions{Enabled: true, Level: pipeline.DefaultCompressLevel, SkipSuffixes: pipeline.DefaultSkipCompressSuffixes}, + }, + { + name: "an out-of-range level is silently clamped, matching real rsync's own documented behavior", + args: []string{"-z", "--compress-level=15"}, + want: pipeline.CompressOptions{Enabled: true, Level: 9, SkipSuffixes: pipeline.DefaultSkipCompressSuffixes}, + }, + { + name: "--skip-compress overrides the default suffix list", + args: []string{"-z", "--skip-compress=foo/bar"}, + want: pipeline.CompressOptions{Enabled: true, Level: pipeline.DefaultCompressLevel, SkipSuffixes: []string{"foo", "bar"}}, + }, + { + name: "--skip-compress=\"\" explicitly means skip nothing, not \"unset\"", + args: []string{"-z", "--skip-compress="}, + want: pipeline.CompressOptions{Enabled: true, Level: pipeline.DefaultCompressLevel, SkipSuffixes: nil}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cmd, opts := buildCompressTestCmd(t, tt.args) + got := effectiveCompressOptions(cmd, opts) + if got.Enabled != tt.want.Enabled || got.Level != tt.want.Level || len(got.SkipSuffixes) != len(tt.want.SkipSuffixes) { + t.Fatalf("effectiveCompressOptions(%v) = %+v, want %+v", tt.args, got, tt.want) + } + for i := range tt.want.SkipSuffixes { + if got.SkipSuffixes[i] != tt.want.SkipSuffixes[i] { + t.Errorf("effectiveCompressOptions(%v).SkipSuffixes = %v, want %v", tt.args, got.SkipSuffixes, tt.want.SkipSuffixes) + break + } + } + }) + } +} + +// statsFieldForTest extracts the integer following "label: " from a +// --stats output block, e.g. statsFieldForTest(t, out, "Total file size") +// on a line "Total file size: 1,416 bytes" returns 1416 - the CLI +// package's own counterpart to internal/pipeline's statsField, duplicated +// rather than exported across packages for a single test-only helper. +func statsFieldForTest(t *testing.T, output, label string) int64 { + t.Helper() + re := regexp.MustCompile(regexp.QuoteMeta(label) + `: ([\d,]+)`) + m := re.FindStringSubmatch(output) + if m == nil { + t.Fatalf("field %q not found in stats output:\n%s", label, output) + } + n, err := strconv.ParseInt(strings.ReplaceAll(m[1], ",", ""), 10, 64) + if err != nil { + t.Fatalf("parsing field %q value %q: %v", label, m[1], err) + } + return n +} + // TestE2E_LocalToLocal drives the real CLI command - the same code path // an actual user invocation goes through, not just the internal // pipeline functions directly (those are already covered by @@ -224,6 +339,104 @@ func TestE2E_ProgressOutput(t *testing.T) { } } +// TestE2E_CompressReducesBytesSent drives the real CLI command twice +// against the same highly-compressible content - once with --compress, +// once without - and confirms via --stats' own "Total bytes received" +// field (see internal/pipeline's TestReceiver_StatsBytesReceivedReflectCompressedSize +// for why that field, not "sent," carries the file data here) that +// --compress genuinely reduces wire traffic end to end through the real +// command, not just at the internal Sender/Receiver level. +func TestE2E_CompressReducesBytesSent(t *testing.T) { + content := strings.Repeat("the quick brown fox jumps over the lazy dog ", 2000) + + runOnce := func(extraArgs ...string) string { + src, dst := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(src, "big.txt"), content) + + cmd := NewRootCmd() + var out strings.Builder + cmd.SetArgs(append([]string{"-a", "--stats"}, append(extraArgs, src, dst)...)) + cmd.SetOut(&out) + if err := cmd.Execute(); err != nil { + t.Fatalf("Execute returned error: %v", err) + } + + got, err := os.ReadFile(filepath.Join(dst, "big.txt")) + if err != nil { + t.Fatalf("reading synced file: %v", err) + } + if string(got) != content { + t.Fatalf("synced content differs from source (len got=%d, want=%d)", len(got), len(content)) + } + return out.String() + } + + uncompressedOut := runOnce() + compressedOut := runOnce("--compress") + + uncompressedReceived := statsFieldForTest(t, uncompressedOut, "Total bytes received") + compressedReceived := statsFieldForTest(t, compressedOut, "Total bytes received") + if compressedReceived >= uncompressedReceived { + t.Errorf("--compress run received %d bytes, plain run received %d bytes, want --compress meaningfully smaller", compressedReceived, uncompressedReceived) + } +} + +// TestE2E_CompressLevelAndSkipCompressFlagsDoNotBreakTransfer drives the +// real CLI command with --compress-level and --skip-compress together +// and confirms the transfer still completes correctly - these flags must +// never affect correctness, only wire size. +func TestE2E_CompressLevelAndSkipCompressFlagsDoNotBreakTransfer(t *testing.T) { + src, dst := t.TempDir(), t.TempDir() + content := strings.Repeat("compressible content ", 1000) + mustWriteFile(t, filepath.Join(src, "file.bin"), content) + + cmd := NewRootCmd() + cmd.SetArgs([]string{"-a", "--compress", "--compress-level=9", "--skip-compress=bin", src, dst}) + cmd.SetOut(io.Discard) + if err := cmd.Execute(); err != nil { + t.Fatalf("Execute returned error: %v", err) + } + + got, err := os.ReadFile(filepath.Join(dst, "file.bin")) + if err != nil { + t.Fatalf("reading synced file: %v", err) + } + if string(got) != content { + t.Errorf("synced content differs from source (len got=%d, want=%d)", len(got), len(content)) + } +} + +// TestE2E_CompressLevelZeroDisablesCompressionEvenWithDashZ drives the +// real CLI command with -z --compress-level=0 together and confirms the +// transfer still completes correctly - real rsync's own documented +// behavior is that an explicit level of 0 turns compression off even +// when -z was also given (see effectiveCompressOptions' own doc +// comment); this only proves the combination doesn't break anything +// observable from the outside (--stats' "Total bytes received" isn't a +// reliable enough signal at this small a scale to assert "definitely +// uncompressed" against, unlike the bigger-content tests above - +// effectiveCompressOptions_test.go asserts the actual returned +// CompressOptions directly instead). +func TestE2E_CompressLevelZeroDisablesCompressionEvenWithDashZ(t *testing.T) { + src, dst := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(src, "file.txt"), "some content") + + cmd := NewRootCmd() + cmd.SetArgs([]string{"-a", "-z", "--compress-level=0", src, dst}) + cmd.SetOut(io.Discard) + if err := cmd.Execute(); err != nil { + t.Fatalf("Execute returned error: %v", err) + } + + got, err := os.ReadFile(filepath.Join(dst, "file.txt")) + if err != nil { + t.Fatalf("reading synced file: %v", err) + } + if string(got) != "some content" { + t.Errorf("synced content = %q, want %q", got, "some content") + } +} + // TestE2E_HardLinksPreservedWithFlag drives the real CLI command with // -H/--hard-links and confirms two hard-linked source files arrive at // the destination still hard-linked to each other (os.SameFile), not diff --git a/internal/daemon/client.go b/internal/daemon/client.go index cb0fcbe..f75f986 100644 --- a/internal/daemon/client.go +++ b/internal/daemon/client.go @@ -21,9 +21,9 @@ import ( // module must be non-empty: DialClient runs a transfer, not a listing - // callers that want to list a daemon's modules should use DialGreeting // directly with an empty module instead. See DialModule's own doc -// comment for exactly what ropts does and doesn't reach on each +// comment for exactly what ropts and copts do and don't reach on each // direction. -func DialClient(nc net.Conn, module, user string, password PasswordFunc, direction Direction, localPath string, rules []sync.Rule, walkOpts sync.WalkOptions, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions) error { +func DialClient(nc net.Conn, module, user string, password PasswordFunc, direction Direction, localPath string, rules []sync.Rule, walkOpts sync.WalkOptions, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions, copts pipeline.CompressOptions) error { if module == "" { return fmt.Errorf("DialClient requires a module name") } @@ -36,5 +36,5 @@ func DialClient(nc net.Conn, module, user string, password PasswordFunc, directi if err := DialAuth(c, user, password); err != nil { return err } - return DialModule(c, direction, localPath, rules, walkOpts, attrOpts, ropts) + return DialModule(c, direction, localPath, rules, walkOpts, attrOpts, ropts, copts) } diff --git a/internal/daemon/client_test.go b/internal/daemon/client_test.go index 0f5ebf2..5aedc4d 100644 --- a/internal/daemon/client_test.go +++ b/internal/daemon/client_test.go @@ -32,7 +32,7 @@ func TestDialClient_DownloadOverRealTCP(t *testing.T) { nc := dialRawConn(t, addr) dest := t.TempDir() - err := DialClient(nc, "public", "", StaticPassword(""), DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}) + err := DialClient(nc, "public", "", StaticPassword(""), DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}) if err != nil { t.Fatalf("DialClient returned error: %v", err) } @@ -66,7 +66,7 @@ func TestDialClient_UploadOverRealTCP(t *testing.T) { t.Fatalf("compiling empty rule set: %v", err) } - err = DialClient(nc, "incoming", "alice", StaticPassword("hunter2"), DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}) + err = DialClient(nc, "incoming", "alice", StaticPassword("hunter2"), DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}) if err != nil { t.Fatalf("DialClient returned error: %v", err) } @@ -106,7 +106,7 @@ func TestDialClient_PasswordFuncNotCalledForAnonymousModule(t *testing.T) { }) dest := t.TempDir() - err := DialClient(nc, "public", "", poisonedPassword, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}) + err := DialClient(nc, "public", "", poisonedPassword, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}) if err != nil { t.Fatalf("DialClient returned error: %v", err) } @@ -120,7 +120,7 @@ func TestDialClient_RejectsEmptyModule(t *testing.T) { addr, _ := startTestDaemon(t, cfg) nc := dialRawConn(t, addr) - err := DialClient(nc, "", "", StaticPassword(""), DirectionGet, t.TempDir(), nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}) + err := DialClient(nc, "", "", StaticPassword(""), DirectionGet, t.TempDir(), nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}) if err == nil { t.Fatalf("DialClient with an empty module returned nil error, want an error") } diff --git a/internal/daemon/server_test.go b/internal/daemon/server_test.go index a168739..4a2fe3b 100644 --- a/internal/daemon/server_test.go +++ b/internal/daemon/server_test.go @@ -60,7 +60,7 @@ func TestDaemon_RealTCP_AnonymousDownload(t *testing.T) { t.Fatalf("DialAuth: %v", err) } dest := t.TempDir() - if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}); err != nil { + if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -102,7 +102,7 @@ func TestDaemon_RealTCP_AuthenticatedUpload(t *testing.T) { if err != nil { t.Fatalf("compiling empty rule set: %v", err) } - if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}); err != nil { + if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -193,7 +193,7 @@ func TestDaemon_RealTCP_DryRunPutMakesNoChanges(t *testing.T) { t.Fatalf("compiling empty rule set: %v", err) } ropts := pipeline.ReceiverOptions{DryRun: true} - if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, ropts); err != nil { + if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, ropts, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -234,7 +234,7 @@ func TestDaemon_RealTCP_DryRunGetMakesNoChanges(t *testing.T) { dest := t.TempDir() ropts := pipeline.ReceiverOptions{DryRun: true} - if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, ropts); err != nil { + if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, ropts, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -275,7 +275,7 @@ func TestDaemon_RealTCP_StatsWorkForGet(t *testing.T) { dest := t.TempDir() var out bytes.Buffer ropts := pipeline.ReceiverOptions{Stats: true, Output: &out} - if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, ropts); err != nil { + if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, ropts, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -323,7 +323,7 @@ func TestDaemon_RealTCP_PutIgnoresProgressAndStatsButStillWorks(t *testing.T) { // ignored for this direction rather than causing an error. var clientSideOutput bytes.Buffer ropts := pipeline.ReceiverOptions{Progress: true, Stats: true, Output: &clientSideOutput} - if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, ropts); err != nil { + if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, ropts, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule: %v", err) } @@ -346,3 +346,53 @@ func TestDaemon_RealTCP_PutIgnoresProgressAndStatsButStillWorks(t *testing.T) { t.Errorf("daemon error log = %q, want empty", errLog.String()) } } + +// TestDaemon_RealTCP_PutWithCompressUploadsCorrectly is SC-9's real, +// over-the-wire proof for the daemon transport: DirectionPut runs +// pipeline.Sender on the client side (see DialModule's own doc comment), +// exactly where --compress/-z's decision belongs, so this drives that +// same client-side Sender with CompressOptions.Enabled against a real +// TCP daemon connection and confirms the upload still arrives byte- +// correct - the server's Receiver only ever reacts to each deltaMessage's +// own Compressed marker, needing no daemon-protocol change at all. +func TestDaemon_RealTCP_PutWithCompressUploadsCorrectly(t *testing.T) { + modRoot := t.TempDir() + cfg := &Config{Modules: map[string]Module{ + "incoming": {Name: "incoming", Path: modRoot, ReadOnly: false}, + }} + addr, errLog := startTestDaemon(t, cfg) + + client := dialTestDaemon(t, addr) + if _, err := DialGreeting(client, "incoming"); err != nil { + t.Fatalf("DialGreeting: %v", err) + } + if err := DialAuth(client, "", StaticPassword("")); err != nil { + t.Fatalf("DialAuth: %v", err) + } + + src := t.TempDir() + content := "compressible daemon upload content, repeated. " + + "compressible daemon upload content, repeated. " + + "compressible daemon upload content, repeated." + mustWriteFile(t, filepath.Join(src, "upload.txt"), content) + rules, err := sync.CompileRules(nil) + if err != nil { + t.Fatalf("compiling empty rule set: %v", err) + } + + copts := pipeline.CompressOptions{Enabled: true, Level: pipeline.DefaultCompressLevel} + if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, copts); err != nil { + t.Fatalf("DialModule: %v", err) + } + + got, err := os.ReadFile(filepath.Join(modRoot, "upload.txt")) + if err != nil { + t.Fatalf("reading uploaded file from module: %v", err) + } + if string(got) != content { + t.Errorf("uploaded content = %q, want %q", got, content) + } + if errLog.Len() != 0 { + t.Errorf("daemon error log = %q, want empty", errLog.String()) + } +} diff --git a/internal/daemon/session.go b/internal/daemon/session.go index 47b246f..02620f8 100644 --- a/internal/daemon/session.go +++ b/internal/daemon/session.go @@ -134,7 +134,13 @@ func ServeModule(c *conn, m Module) error { if err != nil { return fmt.Errorf("compiling module %q exclude rules: %w", m.Name, err) } - if err := pipeline.Sender(c, m.Path, sync.WalkOptions{Recursive: true}, rules, moduleAttrOptions().HardLinks); err != nil { + // pipeline.CompressOptions{} (disabled): a module download has no + // CLI wiring at all yet (see runSync's own "pulling... is not yet + // supported" restriction), so there is no client-facing --compress + // flag that could reach this server-side Sender call in the first + // place - consistent with that existing scope boundary, not a new + // gap. See the README's Compression section. + if err := pipeline.Sender(c, m.Path, sync.WalkOptions{Recursive: true}, rules, moduleAttrOptions().HardLinks, pipeline.CompressOptions{}); err != nil { return err } return waitForTransferDone(c) @@ -181,7 +187,14 @@ func ServeModule(c *conn, m Module) error { // silently unusable for a DirectionPut, since the daemon protocol has no // channel to carry that reporting text back from the server - see // ServeModule's own comment on the same limitation. -func DialModule(c *conn, direction Direction, localPath string, rules []sync.Rule, walkOpts sync.WalkOptions, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions) error { +// +// copts is the DirectionPut mirror image of that same asymmetry: it only +// matters for that direction's client-side Sender call (--compress/-z is +// entirely a sending-side decision, see pipeline.CompressOptions' own +// doc comment) and is simply unused for DirectionGet, where the +// server-side Sender that would consult it has no CLI wiring at all yet +// (see ServeModule's own comment on that same boundary). +func DialModule(c *conn, direction Direction, localPath string, rules []sync.Rule, walkOpts sync.WalkOptions, attrOpts sync.AttrOptions, ropts pipeline.ReceiverOptions, copts pipeline.CompressOptions) error { directionLine := string(direction) if direction == DirectionPut && ropts.DryRun { directionLine += " " + dryRunToken @@ -205,7 +218,7 @@ func DialModule(c *conn, direction Direction, localPath string, rules []sync.Rul } return writeLine(c.w, transferDone) case DirectionPut: - if err := pipeline.Sender(c, localPath, walkOpts, rules, attrOpts.HardLinks); err != nil { + if err := pipeline.Sender(c, localPath, walkOpts, rules, attrOpts.HardLinks, copts); err != nil { return err } return waitForTransferDone(c) diff --git a/internal/daemon/session_test.go b/internal/daemon/session_test.go index f2c8159..502ea5e 100644 --- a/internal/daemon/session_test.go +++ b/internal/daemon/session_test.go @@ -41,7 +41,7 @@ func TestServeModule_GetDownloadsFiles(t *testing.T) { go func() { serverErrCh <- ServeModule(server, m) }() dest := t.TempDir() - if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{Perms: true}, pipeline.ReceiverOptions{}); err != nil { + if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{Perms: true}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule returned error: %v", err) } if err := <-serverErrCh; err != nil { @@ -76,7 +76,7 @@ func TestServeModule_GetHonorsModuleExclude(t *testing.T) { go func() { serverErrCh <- ServeModule(server, m) }() dest := t.TempDir() - if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}); err != nil { + if err := DialModule(client, DirectionGet, dest, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule returned error: %v", err) } if err := <-serverErrCh; err != nil { @@ -102,7 +102,7 @@ func TestServeModule_PutToReadOnlyModuleFails(t *testing.T) { src := t.TempDir() mustWriteFile(t, filepath.Join(src, "upload.txt"), "trying to push this up") - dialErr := DialModule(client, DirectionPut, src, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}) + dialErr := DialModule(client, DirectionPut, src, nil, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}) if dialErr == nil { t.Fatalf("DialModule against a read-only module returned nil error, want an error") } @@ -129,7 +129,7 @@ func TestServeModule_PutToWritableModuleSucceeds(t *testing.T) { if err != nil { t.Fatalf("compiling empty rule set: %v", err) } - if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}); err != nil { + if err := DialModule(client, DirectionPut, src, rules, sync.WalkOptions{}, sync.AttrOptions{}, pipeline.ReceiverOptions{}, pipeline.CompressOptions{}); err != nil { t.Fatalf("DialModule returned error: %v", err) } if err := <-serverErrCh; err != nil { diff --git a/internal/pipeline/compress.go b/internal/pipeline/compress.go new file mode 100644 index 0000000..00537ec --- /dev/null +++ b/internal/pipeline/compress.go @@ -0,0 +1,177 @@ +package pipeline + +import ( + "bytes" + "compress/zlib" + "fmt" + "io" + "path/filepath" + "strings" +) + +// DefaultCompressLevel is real rsync's own zlib default, verified against +// upstream's token.c (init_compression_level's def_level for the zlib/ +// zlibx choice) rather than assumed. +const DefaultCompressLevel = 6 + +// ClampCompressLevel mirrors real rsync's own --compress-level handling +// for zlib compression (token.c's init_compression_level), verified +// against upstream source and rsync.1's own documented wording rather +// than guessed: 0 is a distinct "off" sentinel, not clamped up into +// range; -1 (zlib.DefaultCompression) explicitly means "use the default +// level" (6); anything else out of [1, 9] is silently limited into range +// ("If you specify a too-large or too-small value, the number is +// silently limited to a valid value" - rsync.1's own wording). +func ClampCompressLevel(level int) int { + switch { + case level == zlib.NoCompression: // 0: explicit "off" + return 0 + case level == zlib.DefaultCompression: // -1: "use the default" + return DefaultCompressLevel + case level < zlib.BestSpeed: // any other value below 1 + return zlib.BestSpeed + case level > zlib.BestCompression: // above 9 + return zlib.BestCompression + default: + return level + } +} + +// CompressOptions governs whether/how Sender compresses each regular +// file's literal delta data (--compress/-z) before sending it - see the +// README's Compression section. It is consulted only by Sender: Receiver +// needs no compression options of its own at all, since each +// deltaMessage's own Compressed marker (messages.go) already says +// whether its literal data needs decompressing first - a purely +// data-driven decision on that side, not a policy one. +type CompressOptions struct { + Enabled bool + // Level is a zlib compression level from 1 (fastest) to 9 (smallest), + // meaningful only when Enabled - see ClampCompressLevel, which every + // caller that constructs a CompressOptions with Enabled: true is + // expected to have already run Level through. + Level int + // SkipSuffixes is a lowercase, dot-free list of file suffixes (e.g. + // "gz", "jpg") to send uncompressed even when Enabled - real rsync's + // own --skip-compress default list (DefaultSkipCompressSuffixes) or a + // caller override (see ParseSkipCompressList). Nil/empty means "skip + // nothing," matching real rsync's own documented meaning of an empty + // --skip-compress=LIST. + SkipSuffixes []string +} + +// DefaultSkipCompressSuffixes is real rsync's own built-in --skip-compress +// suffix list, copied verbatim from rsync.1.md's own documented default +// (the same list default-dont-compress.h is generated from) rather than +// invented - files with one of these suffixes are already compressed +// formats where running zlib over them again wastes CPU for no size +// benefit. +var DefaultSkipCompressSuffixes = []string{ + "3g2", "3gp", "7z", "aac", "ace", "apk", "avi", "bz2", "deb", "dmg", + "ear", "f4v", "flac", "flv", "gpg", "gz", "iso", "jar", "jpeg", "jpg", + "lrz", "lz", "lz4", "lzma", "lzo", "m1a", "m1v", "m2a", "m2ts", "m2v", + "m4a", "m4b", "m4p", "m4r", "m4v", "mka", "mkv", "mov", "mp1", "mp2", + "mp3", "mp4", "mpa", "mpeg", "mpg", "mpv", "mts", "odb", "odf", "odg", + "odi", "odm", "odp", "ods", "odt", "oga", "ogg", "ogm", "ogv", "ogx", + "opus", "otg", "oth", "otp", "ots", "ott", "oxt", "png", "qt", "rar", + "rpm", "rz", "rzip", "spx", "squashfs", "sxc", "sxd", "sxg", "sxm", + "sxw", "sz", "tbz", "tbz2", "tgz", "tlz", "ts", "txz", "tzo", "vob", + "war", "webm", "webp", "xz", "z", "zip", "zst", +} + +// ParseSkipCompressList parses a real rsync --skip-compress=LIST value: +// suffixes without their leading dot, separated by "/". An empty string +// is a meaningful value in its own right ("skip nothing"), not "unset" - +// see effectiveCompressOptions (internal/cli) for how that distinction +// from "the flag was never given at all" is made. +// +// Real rsync's own LIST grammar also supports bracketed character +// classes inside a suffix (e.g. "mp[34]" for "mp3"/"mp4"); grsync's +// --skip-compress does not, a deliberate, disclosed scope reduction (see +// the README's Compression section) rather than a silent gap - plain +// slash-separated suffixes cover the default list and the overwhelming +// majority of real-world uses. +func ParseSkipCompressList(list string) []string { + if list == "" { + return nil + } + parts := strings.Split(list, "/") + out := make([]string, 0, len(parts)) + for _, p := range parts { + if p == "" { + continue + } + out = append(out, strings.ToLower(p)) + } + return out +} + +// skipCompressSuffix reports whether path's suffix (case-insensitively, +// without its leading dot) appears in skipSuffixes. +func skipCompressSuffix(path string, skipSuffixes []string) bool { + ext := filepath.Ext(path) + if ext == "" { + return false + } + suffix := strings.ToLower(strings.TrimPrefix(ext, ".")) + for _, s := range skipSuffixes { + if s == suffix { + return true + } + } + return false +} + +// compressLiteral zlib-compresses data at level, returning ok == false if +// compression didn't actually help (the result is not smaller than data +// itself) so the caller can fall back to sending it raw. +// +// This check matters more here than it would for real rsync's own zlib +// usage: real rsync keeps one persistent deflate stream open per file, +// so its fixed ~8-byte zlib header/trailer cost is paid once per file no +// matter how many separate literal runs cross the wire. grsync's +// deltaMessage is sent as a single, independent frame per file with no +// persistent compression context to reuse across files - toWireDeltaOps +// already amortizes that overhead across everything within one file by +// compressing the whole concatenated literal stream as a single unit +// rather than op-by-op (see deltaMessage's own doc comment), but a file +// whose total literal data is tiny (a few changed bytes in an otherwise- +// unchanged large file - exactly the case delta transfer exists for) or +// already-incompressible can still legitimately come out larger +// compressed than raw. Falling back per file, only when it doesn't pay +// off, is a small, real improvement over always compressing regardless. +func compressLiteral(data []byte, level int) (compressed []byte, ok bool) { + var buf bytes.Buffer + w, err := zlib.NewWriterLevel(&buf, level) + if err != nil { + // ClampCompressLevel guarantees level is 1-9, which zlib always + // accepts - this should be unreachable, but treating any error as + // "just send raw" is safe (a pure optimization, never required + // for correctness) rather than propagating a hard failure for it. + return nil, false + } + if _, err := w.Write(data); err != nil { + return nil, false + } + if err := w.Close(); err != nil { + return nil, false + } + if buf.Len() >= len(data) { + return nil, false + } + return buf.Bytes(), true +} + +// decompressLiteral reverses compressLiteral. +func decompressLiteral(data []byte) ([]byte, error) { + r, err := zlib.NewReader(bytes.NewReader(data)) + if err != nil { + return nil, fmt.Errorf("opening zlib reader: %w", err) + } + defer func() { _ = r.Close() }() + out, err := io.ReadAll(r) + if err != nil { + return nil, fmt.Errorf("reading decompressed data: %w", err) + } + return out, nil +} diff --git a/internal/pipeline/compress_test.go b/internal/pipeline/compress_test.go new file mode 100644 index 0000000..b86ee6e --- /dev/null +++ b/internal/pipeline/compress_test.go @@ -0,0 +1,113 @@ +package pipeline + +import ( + "bytes" + "strings" + "testing" +) + +func TestClampCompressLevel(t *testing.T) { + tests := []struct { + in int + want int + }{ + {0, 0}, // explicit "off" + {-1, 6}, // zlib.DefaultCompression -> real rsync's own default + {1, 1}, // BestSpeed, unchanged + {6, 6}, // already the default, unchanged + {9, 9}, // BestCompression, unchanged + {10, 9}, // too large, silently limited (rsync.1's own wording) + {999, 9}, // wildly too large, still limited to 9 + {-5, 1}, // any other negative, limited up to the minimum + {3, 3}, + } + for _, tt := range tests { + if got := ClampCompressLevel(tt.in); got != tt.want { + t.Errorf("ClampCompressLevel(%d) = %d, want %d", tt.in, got, tt.want) + } + } +} + +func TestParseSkipCompressList(t *testing.T) { + tests := []struct { + in string + want []string + }{ + {"", nil}, + {"gz", []string{"gz"}}, + {"gz/jpg/mp3", []string{"gz", "jpg", "mp3"}}, + {"GZ/Jpg", []string{"gz", "jpg"}}, // lowercased, matching real rsync's own add_nocompress_suffixes + {"gz//jpg", []string{"gz", "jpg"}}, // empty segments dropped + } + for _, tt := range tests { + got := ParseSkipCompressList(tt.in) + if len(got) != len(tt.want) { + t.Errorf("ParseSkipCompressList(%q) = %v, want %v", tt.in, got, tt.want) + continue + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("ParseSkipCompressList(%q) = %v, want %v", tt.in, got, tt.want) + break + } + } + } +} + +func TestSkipCompressSuffix(t *testing.T) { + suffixes := []string{"gz", "jpg"} + tests := []struct { + path string + want bool + }{ + {"archive.gz", true}, + {"ARCHIVE.GZ", true}, // case-insensitive, matching real rsync's own lowercasing + {"photo.jpg", true}, + {"notes.txt", false}, + {"noextension", false}, + {"dir.gz/file.txt", false}, // the suffix must belong to the final path component + } + for _, tt := range tests { + if got := skipCompressSuffix(tt.path, suffixes); got != tt.want { + t.Errorf("skipCompressSuffix(%q, %v) = %v, want %v", tt.path, suffixes, got, tt.want) + } + } +} + +func TestCompressDecompressLiteral_RoundTrip(t *testing.T) { + data := []byte(strings.Repeat("hello compressible world ", 500)) + for level := 1; level <= 9; level++ { + compressed, ok := compressLiteral(data, level) + if !ok { + t.Fatalf("level %d: compressLiteral did not compress genuinely-compressible data", level) + } + if len(compressed) >= len(data) { + t.Errorf("level %d: compressed len %d >= raw len %d, want smaller", level, len(compressed), len(data)) + } + got, err := decompressLiteral(compressed) + if err != nil { + t.Fatalf("level %d: decompressLiteral returned error: %v", level, err) + } + if !bytes.Equal(got, data) { + t.Errorf("level %d: round trip mismatch: got %d bytes, want %d bytes identical to original", level, len(got), len(data)) + } + } +} + +// TestCompressLiteral_FallsBackWhenNotSmaller is the self-review's tiny- +// payload concern made concrete: a handful of bytes can never come out +// smaller after zlib's own ~8-byte header/trailer overhead, so +// compressLiteral must report ok == false rather than silently returning +// something larger than the input. +func TestCompressLiteral_FallsBackWhenNotSmaller(t *testing.T) { + tiny := []byte{1, 2, 3} + if _, ok := compressLiteral(tiny, DefaultCompressLevel); ok { + t.Errorf("compressLiteral(%v) reported ok, want false - 3 bytes can never compress smaller than zlib's own fixed overhead", tiny) + } +} + +func TestCompressLiteral_EmptyInput(t *testing.T) { + if _, ok := compressLiteral(nil, DefaultCompressLevel); ok { + t.Errorf("compressLiteral(nil) reported ok, want false - there is nothing to gain by compressing zero bytes") + } +} diff --git a/internal/pipeline/messages.go b/internal/pipeline/messages.go index 5d72d3d..f87aa17 100644 --- a/internal/pipeline/messages.go +++ b/internal/pipeline/messages.go @@ -41,32 +41,84 @@ const ( type wireDeltaOp struct { Kind deltaOpKind BlockIndex int // valid when Kind == deltaOpKindCopy - Bytes []byte // valid when Kind == deltaOpKindData + Bytes []byte // valid when Kind == deltaOpKindData and the enclosing deltaMessage is NOT compressed + Length int // valid when Kind == deltaOpKindData and the enclosing deltaMessage IS compressed: how many bytes of its decompressed Literal stream belong to this op } -func toWireDeltaOps(ops []sync.DeltaOp) ([]wireDeltaOp, error) { - wire := make([]wireDeltaOp, len(ops)) +// toWireDeltaOps converts ops to their wire form, compressing this +// file's entire literal-data stream as a single zlib unit when copts +// calls for it (see deltaMessage's own doc comment for why whole-file, +// not per-op) - path is only used to check copts.SkipSuffixes, never +// otherwise. CopyOp block indices are never touched: they're plain +// integers, not data, and compressing them would only add overhead for +// nothing. +func toWireDeltaOps(ops []sync.DeltaOp, path string, copts CompressOptions) (wire []wireDeltaOp, compressed bool, literal []byte, err error) { + wire = make([]wireDeltaOp, len(ops)) + + tryCompress := copts.Enabled && !skipCompressSuffix(path, copts.SkipSuffixes) + var concatenated []byte + if tryCompress { + for _, op := range ops { + if d, ok := op.(sync.DataOp); ok { + concatenated = append(concatenated, d.Bytes...) + } + } + if len(concatenated) > 0 { + if c, ok := compressLiteral(concatenated, copts.Level); ok { + literal = c + compressed = true + } + } + } + for i, op := range ops { switch o := op.(type) { case sync.CopyOp: wire[i] = wireDeltaOp{Kind: deltaOpKindCopy, BlockIndex: o.BlockIndex} case sync.DataOp: - wire[i] = wireDeltaOp{Kind: deltaOpKindData, Bytes: o.Bytes} + if compressed { + wire[i] = wireDeltaOp{Kind: deltaOpKindData, Length: len(o.Bytes)} + } else { + wire[i] = wireDeltaOp{Kind: deltaOpKindData, Bytes: o.Bytes} + } default: - return nil, fmt.Errorf("op %d: unknown DeltaOp type %T", i, op) + return nil, false, nil, fmt.Errorf("op %d: unknown DeltaOp type %T", i, op) } } - return wire, nil + return wire, compressed, literal, nil } -func fromWireDeltaOps(wire []wireDeltaOp) ([]sync.DeltaOp, error) { +// fromWireDeltaOps reverses toWireDeltaOps: when compressed is true, it +// decompresses literal once and re-slices it back into each op's own +// bytes using the Length each wireDeltaOp carried; otherwise each op's +// Bytes is used directly, exactly as before compression existed. +func fromWireDeltaOps(wire []wireDeltaOp, compressed bool, literal []byte) ([]sync.DeltaOp, error) { + var decompressed []byte + if compressed { + var err error + decompressed, err = decompressLiteral(literal) + if err != nil { + return nil, fmt.Errorf("decompressing literal data: %w", err) + } + } + ops := make([]sync.DeltaOp, len(wire)) + pos := 0 for i, w := range wire { switch w.Kind { case deltaOpKindCopy: ops[i] = sync.CopyOp{BlockIndex: w.BlockIndex} case deltaOpKindData: - ops[i] = sync.DataOp{Bytes: w.Bytes} + if !compressed { + ops[i] = sync.DataOp{Bytes: w.Bytes} + continue + } + end := pos + w.Length + if w.Length < 0 || end > len(decompressed) { + return nil, fmt.Errorf("op %d: decompressed literal stream too short (want %d more bytes at offset %d, have %d total)", i, w.Length, pos, len(decompressed)) + } + ops[i] = sync.DataOp{Bytes: decompressed[pos:end]} + pos = end default: return nil, fmt.Errorf("op %d: unknown wire delta op kind %d", i, w.Kind) } @@ -88,9 +140,24 @@ type signatureMessage struct { // deltaMessage is FrameDelta's payload: one regular file's delta ops, // tagged with its Path for the same reason as signatureMessage. +// +// Literal holds every DataOp's bytes for this file, zlib-compressed +// together as a single stream when Compressed is true - each op's own +// wireDeltaOp.Bytes is left empty in that case, and its Length instead +// says how many of Literal's decompressed bytes are its (see +// toWireDeltaOps/fromWireDeltaOps). Compressing the whole file's literal +// data as one unit, rather than op-by-op, amortizes zlib's fixed ~8-byte +// header/trailer overhead across the entire file instead of paying it +// again for every small literal run a scattered-changes file can +// produce - see compressLiteral's own doc comment. When Compressed is +// false, Literal is unused (nil) and every op carries its own Bytes +// directly, exactly the wire shape this type had before --compress +// existed. type deltaMessage struct { - Path string - Ops []wireDeltaOp + Path string + Ops []wireDeltaOp + Compressed bool + Literal []byte } func encodeGob(v any) ([]byte, error) { @@ -178,12 +245,12 @@ func recvSignature(r io.Reader) (signatureMessage, error) { return msg, nil } -func sendDelta(w io.Writer, path string, ops []sync.DeltaOp) error { - wire, err := toWireDeltaOps(ops) +func sendDelta(w io.Writer, path string, ops []sync.DeltaOp, copts CompressOptions) error { + wire, compressed, literal, err := toWireDeltaOps(ops, path, copts) if err != nil { return fmt.Errorf("converting delta for %q: %w", path, err) } - payload, err := encodeGob(deltaMessage{Path: path, Ops: wire}) + payload, err := encodeGob(deltaMessage{Path: path, Ops: wire, Compressed: compressed, Literal: literal}) if err != nil { return fmt.Errorf("encoding delta for %q: %w", path, err) } @@ -199,7 +266,7 @@ func recvDelta(r io.Reader) (path string, ops []sync.DeltaOp, err error) { if err := decodeGob(f.Payload, &msg); err != nil { return "", nil, fmt.Errorf("decoding delta: %w", err) } - ops, err = fromWireDeltaOps(msg.Ops) + ops, err = fromWireDeltaOps(msg.Ops, msg.Compressed, msg.Literal) if err != nil { return "", nil, fmt.Errorf("converting delta for %q: %w", msg.Path, err) } diff --git a/internal/pipeline/messages_test.go b/internal/pipeline/messages_test.go index 3fae962..6454f47 100644 --- a/internal/pipeline/messages_test.go +++ b/internal/pipeline/messages_test.go @@ -3,6 +3,7 @@ package pipeline import ( "bytes" "io/fs" + "strings" "testing" "time" @@ -110,7 +111,7 @@ func TestDeltaRoundTrip(t *testing.T) { } var buf bytes.Buffer - if err := sendDelta(&buf, "some/file.txt", ops); err != nil { + if err := sendDelta(&buf, "some/file.txt", ops, CompressOptions{}); err != nil { t.Fatalf("sendDelta returned error: %v", err) } gotPath, gotOps, err := recvDelta(&buf) @@ -140,6 +141,127 @@ func TestDeltaRoundTrip(t *testing.T) { } } +// TestDeltaRoundTrip_Compressed is SC-9's core wire-format proof: sending +// a delta with compression enabled must produce a smaller frame than the +// same ops sent uncompressed, and recvDelta on the compressed side must +// still reconstruct byte-identical ops - proving toWireDeltaOps/ +// fromWireDeltaOps' compress-once-per-file, re-slice-by-Length design +// (see deltaMessage's own doc comment) round-trips correctly, not just +// that compression was attempted. +func TestDeltaRoundTrip_Compressed(t *testing.T) { + literal := []byte(strings.Repeat("compressible literal data ", 200)) + ops := []sync.DeltaOp{ + sync.CopyOp{BlockIndex: 3}, + sync.DataOp{Bytes: literal[:len(literal)/2]}, + sync.CopyOp{BlockIndex: 1}, + sync.DataOp{Bytes: literal[len(literal)/2:]}, + } + + var uncompressed bytes.Buffer + if err := sendDelta(&uncompressed, "file.txt", ops, CompressOptions{}); err != nil { + t.Fatalf("sendDelta (uncompressed) returned error: %v", err) + } + + var compressed bytes.Buffer + copts := CompressOptions{Enabled: true, Level: DefaultCompressLevel} + if err := sendDelta(&compressed, "file.txt", ops, copts); err != nil { + t.Fatalf("sendDelta (compressed) returned error: %v", err) + } + + if compressed.Len() >= uncompressed.Len() { + t.Errorf("compressed frame = %d bytes, uncompressed = %d bytes, want compressed smaller", compressed.Len(), uncompressed.Len()) + } + + gotPath, gotOps, err := recvDelta(&compressed) + if err != nil { + t.Fatalf("recvDelta returned error: %v", err) + } + if gotPath != "file.txt" { + t.Errorf("path = %q, want %q", gotPath, "file.txt") + } + if len(gotOps) != len(ops) { + t.Fatalf("got %d ops, want %d", len(gotOps), len(ops)) + } + for i := range ops { + switch want := ops[i].(type) { + case sync.CopyOp: + got, ok := gotOps[i].(sync.CopyOp) + if !ok || got != want { + t.Errorf("op %d = %+v, want %+v", i, gotOps[i], want) + } + case sync.DataOp: + got, ok := gotOps[i].(sync.DataOp) + if !ok || !bytes.Equal(got.Bytes, want.Bytes) { + t.Errorf("op %d length = %d, want %d, equal=%v", i, len(got.Bytes), len(want.Bytes), ok && bytes.Equal(got.Bytes, want.Bytes)) + } + } + } +} + +// TestDeltaRoundTrip_SkipCompressSuffixIsGenuinelyUncompressed is the +// ticket's explicit "inspect actual bytes sent, not just trust the flag +// was read" requirement: it decodes the raw gob frame directly (the same +// way recvDelta itself does internally) and asserts deltaMessage.Compressed +// is false and every op still carries its own literal Bytes, for a path +// whose suffix is in copts.SkipSuffixes - even though copts.Enabled is +// true and the literal data is highly compressible. +func TestDeltaRoundTrip_SkipCompressSuffixIsGenuinelyUncompressed(t *testing.T) { + literal := []byte(strings.Repeat("z", 5000)) // trivially compressible + ops := []sync.DeltaOp{sync.DataOp{Bytes: literal}} + copts := CompressOptions{Enabled: true, Level: DefaultCompressLevel, SkipSuffixes: []string{"bin"}} + + var buf bytes.Buffer + if err := sendDelta(&buf, "archive.bin", ops, copts); err != nil { + t.Fatalf("sendDelta returned error: %v", err) + } + + f, err := transport.ReadFrame(&buf) + if err != nil { + t.Fatalf("ReadFrame returned error: %v", err) + } + var msg deltaMessage + if err := decodeGob(f.Payload, &msg); err != nil { + t.Fatalf("decodeGob returned error: %v", err) + } + + if msg.Compressed { + t.Errorf("deltaMessage.Compressed = true for a .bin path in SkipSuffixes, want false") + } + if msg.Literal != nil { + t.Errorf("deltaMessage.Literal = %d bytes, want nil (unused) when not compressed", len(msg.Literal)) + } + if len(msg.Ops) != 1 || !bytes.Equal(msg.Ops[0].Bytes, literal) { + t.Errorf("op 0 did not carry its own literal Bytes directly despite Compressed being false") + } +} + +// TestDeltaRoundTrip_DisabledIsGenuinelyUncompressed is +// TestDeltaRoundTrip_SkipCompressSuffixIsGenuinelyUncompressed's +// counterpart for CompressOptions{} (the zero value, --compress not +// given at all): the wire frame must be byte-identical in shape to the +// pre-SC-9 format, not merely "the same size by coincidence." +func TestDeltaRoundTrip_DisabledIsGenuinelyUncompressed(t *testing.T) { + ops := []sync.DeltaOp{sync.DataOp{Bytes: []byte(strings.Repeat("y", 5000))}} + + var buf bytes.Buffer + if err := sendDelta(&buf, "file.dat", ops, CompressOptions{}); err != nil { + t.Fatalf("sendDelta returned error: %v", err) + } + + f, err := transport.ReadFrame(&buf) + if err != nil { + t.Fatalf("ReadFrame returned error: %v", err) + } + var msg deltaMessage + if err := decodeGob(f.Payload, &msg); err != nil { + t.Fatalf("decodeGob returned error: %v", err) + } + + if msg.Compressed { + t.Errorf("deltaMessage.Compressed = true with CompressOptions{}, want false") + } +} + func TestReadTypedFrame_TranslatesFrameErrorToGoError(t *testing.T) { var buf bytes.Buffer if err := transport.WriteFrame(&buf, transport.Frame{Type: transport.FrameError, Payload: []byte("remote blew up")}); err != nil { diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index 2263c2b..3441ac0 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -28,7 +28,7 @@ func runSenderReceiver(t *testing.T, src, dest string, walkOpts sync.WalkOptions receiver := pipeReadWriter{Reader: receiverReadsFromSender, Writer: receiverWritesToSender} senderErrCh := make(chan error, 1) - go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks) }() + go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks, CompressOptions{}) }() receiverErrCh := make(chan error, 1) go func() { receiverErrCh <- Receiver(receiver, dest, attrOpts, ReceiverOptions{}) }() @@ -363,7 +363,7 @@ func runSenderReceiverWithOptions(t *testing.T, src, dest string, walkOpts sync. receiver := pipeReadWriter{Reader: receiverReadsFromSender, Writer: receiverWritesToSender} senderErrCh := make(chan error, 1) - go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks) }() + go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks, CompressOptions{}) }() receiverErrCh := make(chan error, 1) go func() { receiverErrCh <- Receiver(receiver, dest, attrOpts, ropts) }() @@ -386,6 +386,303 @@ func runSenderReceiverWithOptions(t *testing.T, src, dest string, walkOpts sync. } } +// runSenderReceiverWithCompressOptions is runSenderReceiverWithOptions's +// counterpart for tests that need control over the sender's compression +// behavior - kept separate for the same reason runSenderReceiverWithOptions +// itself was (see its own doc comment): none of the many existing +// runSenderReceiver/runSenderReceiverWithOptions callers care about +// compression, so Sender's CompressOptions parameter stays defaulted to +// CompressOptions{} (disabled) in those two instead of forcing every +// existing call site to pass one. +// +// Returns the number of bytes Sender actually wrote to the connection +// (via the same countingReadWriter Stats itself uses - see stats.go), so +// callers can compare compressed vs. uncompressed wire size against real +// bytes sent, not just trust that the flag was read. +func runSenderReceiverWithCompressOptions(t *testing.T, src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, attrOpts sync.AttrOptions, ropts ReceiverOptions, copts CompressOptions) (bytesWritten int64) { + t.Helper() + + senderReadsFromReceiver, receiverWritesToSender := io.Pipe() + receiverReadsFromSender, senderWritesToReceiver := io.Pipe() + + sender := &countingReadWriter{rw: pipeReadWriter{Reader: senderReadsFromReceiver, Writer: senderWritesToReceiver}} + receiver := pipeReadWriter{Reader: receiverReadsFromSender, Writer: receiverWritesToSender} + + senderErrCh := make(chan error, 1) + go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks, copts) }() + + receiverErrCh := make(chan error, 1) + go func() { receiverErrCh <- Receiver(receiver, dest, attrOpts, ropts) }() + + select { + case err := <-receiverErrCh: + if err != nil { + t.Fatalf("Receiver returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Receiver did not complete within 10s") + } + select { + case err := <-senderErrCh: + if err != nil { + t.Fatalf("Sender returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Sender did not complete within 10s") + } + + return sender.written +} + +// TestSenderReceiver_CompressReducesWireBytesForCompressibleContent is +// SC-9's core end-to-end proof: syncing the exact same highly-compressible +// file with --compress enabled must write genuinely fewer bytes to the +// wire than syncing it uncompressed, and the destination content must +// still come out byte-identical to the source either way - compression +// must never be observable in the result, only in the traffic. +func TestSenderReceiver_CompressReducesWireBytesForCompressibleContent(t *testing.T) { + content := strings.Repeat("the quick brown fox jumps over the lazy dog ", 2000) // ~90KB, highly compressible + + uncompressedSrc, uncompressedDest := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(uncompressedSrc, "big.txt"), content) + uncompressedBytes := runSenderReceiverWithCompressOptions(t, uncompressedSrc, uncompressedDest, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, ReceiverOptions{}, CompressOptions{}) + + compressedSrc, compressedDest := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(compressedSrc, "big.txt"), content) + compressedBytes := runSenderReceiverWithCompressOptions(t, compressedSrc, compressedDest, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, ReceiverOptions{}, + CompressOptions{Enabled: true, Level: DefaultCompressLevel}) + + if compressedBytes >= uncompressedBytes { + t.Errorf("compressed transfer wrote %d bytes, uncompressed wrote %d bytes, want compressed meaningfully smaller", compressedBytes, uncompressedBytes) + } + + assertSameContent(t, filepath.Join(uncompressedSrc, "big.txt"), filepath.Join(uncompressedDest, "big.txt")) + assertSameContent(t, filepath.Join(compressedSrc, "big.txt"), filepath.Join(compressedDest, "big.txt")) +} + +// TestSenderReceiver_SkipCompressLeavesMatchingSuffixUncompressed proves +// --skip-compress end to end, not just at the wire-message level +// (TestDeltaRoundTrip_SkipCompressSuffixIsGenuinelyUncompressed already +// covers that): a highly-compressible file whose suffix is skip-listed +// must write essentially the same number of bytes as a fully-disabled +// compression run, while an identical file whose suffix is NOT +// skip-listed, synced in the same run, still compresses. +func TestSenderReceiver_SkipCompressLeavesMatchingSuffixUncompressed(t *testing.T) { + content := strings.Repeat("the quick brown fox jumps over the lazy dog ", 2000) + + src, dest := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(src, "skip.bin"), content) + + uncompressedRun := t.TempDir() + mustWriteFile(t, filepath.Join(uncompressedRun, "skip.bin"), content) + baselineBytes := runSenderReceiverWithCompressOptions(t, uncompressedRun, t.TempDir(), + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, ReceiverOptions{}, CompressOptions{}) + + skipBytes := runSenderReceiverWithCompressOptions(t, src, dest, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, ReceiverOptions{}, + CompressOptions{Enabled: true, Level: DefaultCompressLevel, SkipSuffixes: []string{"bin"}}) + + // Not asserting exact equality - gob framing overhead can vary by a + // handful of bytes for unrelated reasons - but a skipped file must be + // nowhere near as small as a genuinely compressed one would be; the + // compressible-content test above already shows compression more than + // halving a similarly-sized file, so any close-to-baseline result here + // is conclusive that skip-compress actually took effect. + if skipBytes < baselineBytes*9/10 { + t.Errorf("skip-listed file wrote %d bytes, baseline (uncompressed) wrote %d bytes, want them close - skip-compress should have left this file uncompressed", skipBytes, baselineBytes) + } + + assertSameContent(t, filepath.Join(src, "skip.bin"), filepath.Join(dest, "skip.bin")) +} + +// TestReceiver_StatsBytesReceivedReflectCompressedSize confirms Step 5's +// stats convention: real rsync's own "Total bytes sent"/"Total bytes +// received" measure what actually crossed the wire, which - once +// --compress is in play - genuinely is the compressed size, not the +// original file size. stats.go's countingReadWriter already wraps the +// raw connection needing no changes for this (see its own doc comment); +// this test is the proof that holds end to end, not just an inspection +// of the code. +// +// It's specifically "Total bytes received" (not "sent") that carries the +// file's compressible data here: Stats is computed on the Receiver side +// (SC-10's own design - see stats.go), and Receiver receives the file +// list and delta payloads from Sender while only ever sending small +// signature/ack messages back - so the large, compression-sensitive +// traffic flows in the "received" direction from this side's own point +// of view, not "sent". +func TestReceiver_StatsBytesReceivedReflectCompressedSize(t *testing.T) { + content := strings.Repeat("the quick brown fox jumps over the lazy dog ", 2000) + + uncompressedSrc, uncompressedDest := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(uncompressedSrc, "big.txt"), content) + var uncompressedOut bytes.Buffer + runSenderReceiverWithCompressAndReceiverOptions(t, uncompressedSrc, uncompressedDest, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, + ReceiverOptions{Stats: true, Output: &uncompressedOut}, CompressOptions{}) + + compressedSrc, compressedDest := t.TempDir(), t.TempDir() + mustWriteFile(t, filepath.Join(compressedSrc, "big.txt"), content) + var compressedOut bytes.Buffer + runSenderReceiverWithCompressAndReceiverOptions(t, compressedSrc, compressedDest, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, + ReceiverOptions{Stats: true, Output: &compressedOut}, + CompressOptions{Enabled: true, Level: DefaultCompressLevel}) + + uncompressedReceived := statsField(t, uncompressedOut.String(), "Total bytes received") + compressedReceived := statsField(t, compressedOut.String(), "Total bytes received") + + if compressedReceived >= uncompressedReceived { + t.Errorf("--stats reported compressed run received %d bytes, uncompressed run received %d bytes, want compressed smaller - "+ + "stats must reflect actual wire bytes, not original file size", compressedReceived, uncompressedReceived) + } + + // Total file size is a property of the files themselves, unaffected by + // compression - the two runs synced byte-identical content, so this + // field specifically must match despite bytes sent differing. + if got, want := statsField(t, compressedOut.String(), "Total file size"), statsField(t, uncompressedOut.String(), "Total file size"); got != want { + t.Errorf("Total file size = %d with compression, %d without, want equal - compression must not affect this field", got, want) + } +} + +// runSenderReceiverWithCompressAndReceiverOptions combines +// runSenderReceiverWithCompressOptions and runSenderReceiverWithOptions' +// separate concerns (compression behavior and receiver-side reporting) +// for the one test above that needs both at once, rather than growing +// either of those two into a shared do-everything helper every other +// caller would need to pass zero values through. +func runSenderReceiverWithCompressAndReceiverOptions(t *testing.T, src, dest string, walkOpts sync.WalkOptions, rules []sync.Rule, attrOpts sync.AttrOptions, ropts ReceiverOptions, copts CompressOptions) { + t.Helper() + + senderReadsFromReceiver, receiverWritesToSender := io.Pipe() + receiverReadsFromSender, senderWritesToReceiver := io.Pipe() + + sender := pipeReadWriter{Reader: senderReadsFromReceiver, Writer: senderWritesToReceiver} + receiver := pipeReadWriter{Reader: receiverReadsFromSender, Writer: receiverWritesToSender} + + senderErrCh := make(chan error, 1) + go func() { senderErrCh <- Sender(sender, src, walkOpts, rules, attrOpts.HardLinks, copts) }() + + receiverErrCh := make(chan error, 1) + go func() { receiverErrCh <- Receiver(receiver, dest, attrOpts, ropts) }() + + select { + case err := <-receiverErrCh: + if err != nil { + t.Fatalf("Receiver returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Receiver did not complete within 10s") + } + select { + case err := <-senderErrCh: + if err != nil { + t.Fatalf("Sender returned error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Sender did not complete within 10s") + } +} + +// TestSenderReceiver_CompressWorksWithDryRun confirms Step 5's dry-run +// interaction: the signature/delta exchange (including compression of +// the delta's literal data) still needs to run byte-correct for accurate +// itemize output even though nothing is written - compression must not +// break that, and the dry-run destination must still end up completely +// empty. +func TestSenderReceiver_CompressWorksWithDryRun(t *testing.T) { + srcRoot := t.TempDir() + destRoot := t.TempDir() + content := strings.Repeat("compressible dry-run content ", 500) + mustWriteFile(t, filepath.Join(srcRoot, "file.txt"), content) + + var out bytes.Buffer + runSenderReceiverWithCompressAndReceiverOptions(t, srcRoot, destRoot, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, + ReceiverOptions{DryRun: true, Itemize: true, Output: &out}, + CompressOptions{Enabled: true, Level: DefaultCompressLevel}) + + if !strings.Contains(out.String(), "file.txt") { + t.Errorf("dry-run itemize output = %q, want it to mention file.txt", out.String()) + } + + entries, err := os.ReadDir(destRoot) + if err != nil { + t.Fatalf("ReadDir(destRoot): %v", err) + } + if len(entries) != 0 { + t.Errorf("destRoot is not empty after a dry run with --compress: %v", entries) + } +} + +// TestSenderReceiver_CompressWorksWithHardLinks confirms Step 5's +// hard-link interaction: a hard-link group's secondary members never go +// through the signature/delta exchange at all (Sender/Receiver both skip +// them outright), so --compress is moot for them by construction: this +// test's only job is to confirm that skip still holds correctly with +// --compress enabled, i.e. compression didn't somehow reintroduce a +// signature/delta round trip for a secondary member. +func TestSenderReceiver_CompressWorksWithHardLinks(t *testing.T) { + srcRoot := t.TempDir() + destRoot := t.TempDir() + + content := strings.Repeat("shared hard-linked content ", 500) + mustWriteFile(t, filepath.Join(srcRoot, "original.txt"), content) + if err := os.Link(filepath.Join(srcRoot, "original.txt"), filepath.Join(srcRoot, "linked.txt")); err != nil { + t.Skipf("hard link creation unsupported in this environment: %v", err) + } + + runSenderReceiverWithCompressAndReceiverOptions(t, srcRoot, destRoot, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{Perms: true, Times: true, HardLinks: true}, + ReceiverOptions{}, CompressOptions{Enabled: true, Level: DefaultCompressLevel}) + + assertSameContent(t, filepath.Join(srcRoot, "original.txt"), filepath.Join(destRoot, "original.txt")) + assertSameContent(t, filepath.Join(srcRoot, "linked.txt"), filepath.Join(destRoot, "linked.txt")) + + if !sync.HardLinksSupported() { + return + } + originalInfo, err := os.Stat(filepath.Join(destRoot, "original.txt")) + if err != nil { + t.Fatalf("Stat: %v", err) + } + linkedInfo, err := os.Stat(filepath.Join(destRoot, "linked.txt")) + if err != nil { + t.Fatalf("Stat: %v", err) + } + if !os.SameFile(originalInfo, linkedInfo) { + t.Errorf("original.txt and linked.txt are independent files at the destination with --compress enabled, want them still hard-linked") + } +} + +// TestSenderReceiver_CompressWorksWithEmptyAndUnchangedFiles is a +// self-review edge case: toWireDeltaOps only attempts compression when a +// file's concatenated literal data is non-empty (see its own doc +// comment), so a brand-new empty file (zero DataOps, per SC-10's own +// investigation into ApplyDelta's accumulator) and a byte-identical +// unchanged file (all CopyOps, zero DataOps) both take the "nothing to +// compress" path through that same function - this confirms both still +// sync correctly with --compress enabled, not just that compressible +// content does. +func TestSenderReceiver_CompressWorksWithEmptyAndUnchangedFiles(t *testing.T) { + srcRoot := t.TempDir() + destRoot := t.TempDir() + + const unchanged = "byte-identical at both ends, all copy ops" + mustWriteFile(t, filepath.Join(srcRoot, "empty.txt"), "") + mustWriteFile(t, filepath.Join(srcRoot, "unchanged.txt"), unchanged) + mustWriteFile(t, filepath.Join(destRoot, "unchanged.txt"), unchanged) + + runSenderReceiverWithCompressAndReceiverOptions(t, srcRoot, destRoot, + sync.WalkOptions{Recursive: true}, nil, sync.AttrOptions{}, ReceiverOptions{}, + CompressOptions{Enabled: true, Level: DefaultCompressLevel}) + + assertSameContent(t, filepath.Join(srcRoot, "empty.txt"), filepath.Join(destRoot, "empty.txt")) + assertSameContent(t, filepath.Join(srcRoot, "unchanged.txt"), filepath.Join(destRoot, "unchanged.txt")) +} + // buildRichTree creates a source tree exercising every write path // Receiver has: a top-level file, a nested directory with a file inside // it, a symlink, and (best-effort - silently omitted if unsupported in @@ -536,7 +833,7 @@ func TestReceiver_AppliesHardLinksFromReceivedGroups(t *testing.T) { return } ops := sync.GenerateDelta(sigMsg.Sig, []byte(content)) - if err := sendDelta(peerWritesToReceiver, "aaa-primary.txt", ops); err != nil { + if err := sendDelta(peerWritesToReceiver, "aaa-primary.txt", ops, CompressOptions{}); err != nil { peerErrCh <- fmt.Errorf("sending delta: %w", err) return } @@ -551,7 +848,7 @@ func TestReceiver_AppliesHardLinksFromReceivedGroups(t *testing.T) { return } ops = sync.GenerateDelta(sigMsg.Sig, []byte("xxxxx")) - if err := sendDelta(peerWritesToReceiver, "unrelated.txt", ops); err != nil { + if err := sendDelta(peerWritesToReceiver, "unrelated.txt", ops, CompressOptions{}); err != nil { peerErrCh <- fmt.Errorf("sending delta: %w", err) return } @@ -631,7 +928,9 @@ func TestSender_ConnectionDropsMidTransfer(t *testing.T) { }() errCh := make(chan error, 1) - go func() { errCh <- Sender(sender, srcRoot, sync.WalkOptions{Recursive: true}, nil, false) }() + go func() { + errCh <- Sender(sender, srcRoot, sync.WalkOptions{Recursive: true}, nil, false, CompressOptions{}) + }() select { case err := <-errCh: diff --git a/internal/pipeline/sender.go b/internal/pipeline/sender.go index 6329d00..fa6e37b 100644 --- a/internal/pipeline/sender.go +++ b/internal/pipeline/sender.go @@ -33,7 +33,12 @@ import ( // -rlptgoD, no H) - detecting hard links means an extra Lstat per entry, // a cost real rsync doesn't spend unless asked to, so grsync doesn't // either. -func Sender(rw io.ReadWriter, src string, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool) error { +// +// copts governs --compress/-z: entirely a sending-side decision (see +// CompressOptions' own doc comment) - Receiver needs no counterpart +// parameter at all, since decompression is driven purely by what each +// deltaMessage's own Compressed marker says, on every transport. +func Sender(rw io.ReadWriter, src string, walkOpts sync.WalkOptions, rules []sync.Rule, hardLinks bool, copts CompressOptions) error { entries, err := sync.Walk(src, walkOpts) if err != nil { return fmt.Errorf("walking %q: %w", src, err) @@ -88,7 +93,7 @@ func Sender(rw io.ReadWriter, src string, walkOpts sync.WalkOptions, rules []syn } ops := sync.GenerateDelta(sigMsg.Sig, data) - if err := sendDelta(rw, entry.Path, ops); err != nil { + if err := sendDelta(rw, entry.Path, ops, copts); err != nil { return fmt.Errorf("sending delta for %q: %w", entry.Path, err) } } diff --git a/internal/pipeline/ssh_test.go b/internal/pipeline/ssh_test.go index 023b766..b84beca 100644 --- a/internal/pipeline/ssh_test.go +++ b/internal/pipeline/ssh_test.go @@ -80,7 +80,7 @@ func TestSSHLocalhost_SyncRoundTrip(t *testing.T) { sendErrCh := make(chan error, 1) go func() { - sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false) + sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false, CompressOptions{}) }() select { @@ -128,7 +128,7 @@ func TestSSHLocalhost_DryRunMakesNoChanges(t *testing.T) { sendErrCh := make(chan error, 1) go func() { - sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false) + sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false, CompressOptions{}) }() select { @@ -185,7 +185,7 @@ func TestSSHLocalhost_ProgressAndStatsDoNotBreakTheTransfer(t *testing.T) { sendErrCh := make(chan error, 1) go func() { - sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false) + sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false, CompressOptions{}) }() select { @@ -203,3 +203,52 @@ func TestSSHLocalhost_ProgressAndStatsDoNotBreakTheTransfer(t *testing.T) { assertSameContent(t, filepath.Join(src, "big.bin"), filepath.Join(dest, "big.bin")) } + +// TestSSHLocalhost_CompressDoesNotBreakTheTransfer is the real, +// over-the-wire proof that --compress/-z's client-side decision (see +// pipeline.CompressOptions' own doc comment - Sender runs locally here, +// so no remote --server argv change is needed at all, unlike --dry-run/ +// --itemize-changes/--verbose/--progress/--stats) doesn't corrupt or +// interfere with an actual transfer over real SSH: the remote --server +// process needs no compression-related flag on its own command line, +// since its Receiver just reacts to each deltaMessage's own Compressed +// marker. +func TestSSHLocalhost_CompressDoesNotBreakTheTransfer(t *testing.T) { + requireLocalSSHServer(t) + grsyncPath := buildGrsyncBinary(t) + + src := t.TempDir() + dest := t.TempDir() + content := strings.Repeat("compressible ssh transfer content ", 2000) + mustWriteFile(t, filepath.Join(src, "big.txt"), content) + + session, err := transport.Dial("", "", "127.0.0.1", []string{grsyncPath, "--server", dest}) + if err != nil { + t.Fatalf("Dial returned error: %v", err) + } + + if err := transport.Handshake(session); err != nil { + t.Fatalf("Handshake returned error: %v", err) + } + + copts := CompressOptions{Enabled: true, Level: DefaultCompressLevel} + sendErrCh := make(chan error, 1) + go func() { + sendErrCh <- Sender(session, src, sync.WalkOptions{Recursive: true}, nil, false, copts) + }() + + select { + case err := <-sendErrCh: + if err != nil { + t.Fatalf("Sender returned error: %v", err) + } + case <-time.After(20 * time.Second): + t.Fatal("Sender did not complete within 20s") + } + + if err := session.Close(); err != nil { + t.Errorf("Session.Close returned error: %v", err) + } + + assertSameContent(t, filepath.Join(src, "big.txt"), filepath.Join(dest, "big.txt")) +}