diff --git a/cmd/verify/config_precedence_test.go b/cmd/verify/config_precedence_test.go index 0b9a40d2..4e4f3a9e 100644 --- a/cmd/verify/config_precedence_test.go +++ b/cmd/verify/config_precedence_test.go @@ -6,11 +6,12 @@ import ( "os" "path/filepath" "testing" + "time" - "bou.ke/monkey" "go.uber.org/zap" "lvmsync_go/internal/config" + manifestpkg "lvmsync_go/manifest" ) func TestDryRunFlagOverridesEnvAndYAML(t *testing.T) { @@ -21,13 +22,15 @@ func TestDryRunFlagOverridesEnvAndYAML(t *testing.T) { t.Fatalf("write config: %v", err) } t.Setenv("LVMSYNC_DRY_RUN", "true") - r := newStubRunner() called := false - patch := monkey.Patch((*Runner).verifyDevices, func(*Runner, context.Context, *config.Config, string, string, string, *zap.Logger) error { + rebuild := func(_ context.Context, _ string, _ string, _ *zap.Logger, _ time.Duration, _ bool, _ uint32, _ uint32, _ uint32, _ uint32, _ ...manifestpkg.IndexOption) error { + return nil + } + verify := func(_ context.Context, _ *config.Config, _ string, _ string, _ string, _ *zap.Logger) error { called = true return nil - }) - defer patch.Unpatch() + } + r := NewRunnerWithDeps(rebuild, nil, verify) if err := r.Run([]string{"--config", cfgFile, "--dry-run=false", "src", "dst"}, zap.NewNop()); err != nil { t.Fatalf("Run: %v", err) } @@ -44,13 +47,15 @@ func TestDryRunEnvOverridesYAML(t *testing.T) { t.Fatalf("write config: %v", err) } t.Setenv("LVMSYNC_DRY_RUN", "true") - r := newStubRunner() called := false - patch := monkey.Patch((*Runner).verifyDevices, func(*Runner, context.Context, *config.Config, string, string, string, *zap.Logger) error { + rebuild := func(_ context.Context, _ string, _ string, _ *zap.Logger, _ time.Duration, _ bool, _ uint32, _ uint32, _ uint32, _ uint32, _ ...manifestpkg.IndexOption) error { + return nil + } + verify := func(_ context.Context, _ *config.Config, _ string, _ string, _ string, _ *zap.Logger) error { called = true return nil - }) - defer patch.Unpatch() + } + r := NewRunnerWithDeps(rebuild, nil, verify) src := filepath.Join(dir, "src") if err := os.WriteFile(src, []byte("data"), 0o644); err != nil { t.Fatalf("write src: %v", err) diff --git a/cmd/verify/verify.go b/cmd/verify/verify.go index 3d900e32..a1ec87d0 100644 --- a/cmd/verify/verify.go +++ b/cmd/verify/verify.go @@ -31,16 +31,35 @@ import ( // Runner holds dependencies for verify operations. type Runner struct { Rebuild func(ctx context.Context, device, output string, logger *zap.Logger, interval time.Duration, allow bool, cdcMin, cdcAvg, cdcMax, hybrid uint32, opts ...manifestpkg.IndexOption) error + Detect func(ctx context.Context, path string, snap, offline bool, typ, fsFreeze, fsThaw, lvmEsc string, freezeTimeout, thawTimeout time.Duration, esc privilege.Escalator, logger *zap.Logger, runner *device.Runner) (device.Device, error) + verify func(ctx context.Context, cfg *config.Config, src, dst, manifestPath string, logger *zap.Logger) error } // NewRunner returns a Runner with production dependencies. -func NewRunner() *Runner { return &Runner{Rebuild: manifestpkg.Rebuild} } +func NewRunner() *Runner { + r := &Runner{Rebuild: manifestpkg.Rebuild, Detect: device.Detect} + r.verify = r.verifyDevices + return r +} -// NewRunnerWithDeps creates a Runner with custom rebuild function. +// NewRunnerWithDeps creates a Runner with custom dependencies. func NewRunnerWithDeps( rebuild func(ctx context.Context, device, output string, logger *zap.Logger, interval time.Duration, allow bool, cdcMin, cdcAvg, cdcMax, hybrid uint32, opts ...manifestpkg.IndexOption) error, + detect func(ctx context.Context, path string, snap, offline bool, typ, fsFreeze, fsThaw, lvmEsc string, freezeTimeout, thawTimeout time.Duration, esc privilege.Escalator, logger *zap.Logger, runner *device.Runner) (device.Device, error), + verify func(ctx context.Context, cfg *config.Config, src, dst, manifestPath string, logger *zap.Logger) error, ) *Runner { - return &Runner{Rebuild: rebuild} + r := &Runner{Rebuild: rebuild} + if detect != nil { + r.Detect = detect + } else { + r.Detect = device.Detect + } + if verify != nil { + r.verify = verify + } else { + r.verify = r.verifyDevices + } + return r } func init() { @@ -108,7 +127,7 @@ func (r *Runner) Run(args []string, logger *zap.Logger) error { ) return nil } - err = r.verifyDevices(ctx, cfg, remaining[0], remaining[1], cfg.ManifestPath, logger) + err = r.verify(ctx, cfg, remaining[0], remaining[1], cfg.ManifestPath, logger) if cfg.Output == "json" || cfg.Output == "yaml" { out := struct { Verified bool `json:"verified" yaml:"verified"` @@ -149,7 +168,7 @@ func (r *Runner) verifyDevices(ctx context.Context, cfg *config.Config, src, dst } runner := device.NewRunner() - srcDev, err := device.Detect(ctx, src, true, cfg.Offline, cfg.SourceType, cfg.FSFreezeCommand, cfg.FSThawCommand, cfg.LVMEscalation, cfg.FreezeTimeout, cfg.ThawTimeout, esc, logger, runner) + srcDev, err := r.Detect(ctx, src, true, cfg.Offline, cfg.SourceType, cfg.FSFreezeCommand, cfg.FSThawCommand, cfg.LVMEscalation, cfg.FreezeTimeout, cfg.ThawTimeout, esc, logger, runner) if err != nil { return err } @@ -167,7 +186,7 @@ func (r *Runner) verifyDevices(ctx context.Context, cfg *config.Config, src, dst } }() - dstDev, err := device.Detect(ctx, dst, true, cfg.Offline, cfg.DestType, cfg.FSFreezeCommand, cfg.FSThawCommand, cfg.LVMEscalation, cfg.FreezeTimeout, cfg.ThawTimeout, esc, logger, runner) + dstDev, err := r.Detect(ctx, dst, true, cfg.Offline, cfg.DestType, cfg.FSFreezeCommand, cfg.FSThawCommand, cfg.LVMEscalation, cfg.FreezeTimeout, cfg.ThawTimeout, esc, logger, runner) if err != nil { return err } @@ -271,12 +290,16 @@ type blockReader interface { } func verifyWithManifest(cfg *config.Config, devicePath, manifestPath string, logger *zap.Logger) error { + hash, err := digestFunc(cfg) + if err != nil { + return err + } return verifyWithManifestOpen(func(path string, direct, strict bool) (blockReader, error) { return blockio.Open(path, direct, strict) - }, cfg, devicePath, manifestPath, logger) + }, hash, cfg, devicePath, manifestPath, logger) } -func verifyWithManifestOpen(open func(string, bool, bool) (blockReader, error), cfg *config.Config, devicePath, manifestPath string, logger *zap.Logger) error { +func verifyWithManifestOpen(open func(string, bool, bool) (blockReader, error), hash func([]byte) [32]byte, cfg *config.Config, devicePath, manifestPath string, logger *zap.Logger) error { idx, err := manifestpkg.Open(manifestPath) if err != nil { return fmt.Errorf("open manifest: %w", err) @@ -325,10 +348,6 @@ func verifyWithManifestOpen(open func(string, bool, bool) (blockReader, error), tasks := make(chan job, workers) errCh := make(chan error, 1) var mismatches int64 - hash, err := digestFunc(cfg) - if err != nil { - return err - } var wg sync.WaitGroup for i := 0; i < workers; i++ { wg.Add(1) diff --git a/cmd/verify/verify_mismatch_test.go b/cmd/verify/verify_mismatch_test.go index 14d9dce8..c0db4543 100644 --- a/cmd/verify/verify_mismatch_test.go +++ b/cmd/verify/verify_mismatch_test.go @@ -36,7 +36,7 @@ func TestRunLogsMismatchBlock(t *testing.T) { return err } return idx.Close() - }) + }, nil, nil) err := r.Run([]string{"--digest", "blake3", src, dst}, logger) if err == nil { t.Fatalf("expected error") @@ -68,7 +68,7 @@ func TestRunLogsMismatchBlockSHA256(t *testing.T) { return err } return idx.Close() - }) + }, nil, nil) err := r.Run([]string{"--digest", "sha256", src, dst}, logger) if err == nil { t.Fatalf("expected error") diff --git a/cmd/verify/verify_test.go b/cmd/verify/verify_test.go index f8f5d689..a02516f5 100644 --- a/cmd/verify/verify_test.go +++ b/cmd/verify/verify_test.go @@ -12,7 +12,6 @@ import ( "testing" "time" - "bou.ke/monkey" "github.com/zeebo/blake3" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -20,6 +19,7 @@ import ( "gopkg.in/yaml.v3" device "lvmsync_go/device" + "lvmsync_go/internal/blockio" "lvmsync_go/internal/config" privilege "lvmsync_go/internal/privilege" manifestpkg "lvmsync_go/manifest" @@ -54,9 +54,10 @@ func createTestFile(t testing.TB, size int) string { // newStubRunner returns a Runner with a no-op rebuild function. func newStubRunner() *Runner { - return NewRunnerWithDeps(func(_ context.Context, _ string, output string, logger *zap.Logger, interval time.Duration, allow bool, cdcMin, cdcAvg, cdcMax, hybrid uint32, opts ...manifestpkg.IndexOption) error { + rebuild := func(_ context.Context, _ string, _ string, _ *zap.Logger, _ time.Duration, _ bool, _ uint32, _ uint32, _ uint32, _ uint32, _ ...manifestpkg.IndexOption) error { return nil - }) + } + return NewRunnerWithDeps(rebuild, nil, nil) } func createManifest(t testing.TB, file string) { @@ -321,23 +322,23 @@ func TestVerifyWithManifestParallel(t *testing.T) { idx.Close() f.Close() - orig := blake3.Sum256 - patch := monkey.Patch(blake3.Sum256, func(p []byte) [32]byte { + slow := func(p []byte) [32]byte { time.Sleep(50 * time.Millisecond) - return orig(p) - }) - defer patch.Unpatch() - + return blake3.Sum256(p) + } + open := func(path string, direct, strict bool) (blockReader, error) { + return blockio.Open(path, direct, strict) + } cfg := &config.Config{Parallel: 1, ChecksumAlgorithm: "blake3"} start := time.Now() - if err := verifyWithManifest(cfg, src, manifestPath, zap.NewNop()); err != nil { + if err := verifyWithManifestOpen(open, slow, cfg, src, manifestPath, zap.NewNop()); err != nil { t.Fatalf("parallel=1: %v", err) } d1 := time.Since(start) cfg.Parallel = 4 start = time.Now() - if err := verifyWithManifest(cfg, src, manifestPath, zap.NewNop()); err != nil { + if err := verifyWithManifestOpen(open, slow, cfg, src, manifestPath, zap.NewNop()); err != nil { t.Fatalf("parallel=4: %v", err) } d2 := time.Since(start) @@ -378,7 +379,7 @@ func TestVerifyDevicesRebuildsManifest(t *testing.T) { return err } return idx.Close() - }) + }, nil, nil) cfg := &config.Config{ChecksumAlgorithm: "blake3"} if err := r.verifyDevices(context.Background(), cfg, src, dst, "", zap.NewNop()); err != nil { t.Fatalf("verifyDevices: %v", err) @@ -398,10 +399,10 @@ func TestVerifyDevicesTimeout(t *testing.T) { if err := os.WriteFile(dst, []byte("foo"), 0o600); err != nil { t.Fatalf("write dst: %v", err) } - r := NewRunnerWithDeps(func(ctx context.Context, _ string, output string, logger *zap.Logger, interval time.Duration, allow bool, cdcMin, cdcAvg, cdcMax, hybrid uint32, opts ...manifestpkg.IndexOption) error { + r := NewRunnerWithDeps(func(ctx context.Context, _ string, _ string, _ *zap.Logger, _ time.Duration, _ bool, _ uint32, _ uint32, _ uint32, _ uint32, _ ...manifestpkg.IndexOption) error { <-ctx.Done() return ctx.Err() - }) + }, nil, nil) cfg := &config.Config{ManifestTimeout: time.Millisecond, ChecksumAlgorithm: "blake3"} err := r.verifyDevices(context.Background(), cfg, src, dst, "", zap.NewNop()) if !errors.Is(err, context.DeadlineExceeded) { @@ -410,15 +411,16 @@ func TestVerifyDevicesTimeout(t *testing.T) { } func TestVerifyDevicesContextCancelled(t *testing.T) { - r := newStubRunner() ctx, cancel := context.WithCancel(context.Background()) called := make(chan struct{}) - patch := monkey.Patch(device.Detect, func(ctx context.Context, _ string, _ bool, _ bool, _ string, _ string, _ string, _ string, _ time.Duration, _ time.Duration, _ privilege.Escalator, _ *zap.Logger, _ *device.Runner) (device.Device, error) { + detect := func(ctx context.Context, _ string, _ bool, _ bool, _ string, _ string, _ string, _ string, _ time.Duration, _ time.Duration, _ privilege.Escalator, _ *zap.Logger, _ *device.Runner) (device.Device, error) { close(called) <-ctx.Done() return nil, ctx.Err() - }) - defer patch.Unpatch() + } + r := NewRunnerWithDeps(func(_ context.Context, _ string, _ string, _ *zap.Logger, _ time.Duration, _ bool, _ uint32, _ uint32, _ uint32, _ uint32, _ ...manifestpkg.IndexOption) error { + return nil + }, detect, nil) cfg := &config.Config{ChecksumAlgorithm: "blake3"} errCh := make(chan error, 1) go func() { @@ -575,7 +577,7 @@ func TestVerifyWithManifestClosesOnce(t *testing.T) { t.Fatalf("manifest close: %v", err) } cfg := &config.Config{Parallel: 1, ODirect: true, BlockSize: int(blockSize)} - if err := verifyWithManifestOpen(open, cfg, "src", manifestPath, zap.NewNop()); err != nil { + if err := verifyWithManifestOpen(open, blake3.Sum256, cfg, "src", manifestPath, zap.NewNop()); err != nil { t.Fatalf("verifyWithManifestOpen: %v", err) } if c1 != 1 { diff --git a/go.mod b/go.mod index ef72e088..fc1cba23 100644 --- a/go.mod +++ b/go.mod @@ -3,9 +3,9 @@ module lvmsync_go go 1.24.3 require ( - bou.ke/monkey v1.0.2 github.com/bits-and-blooms/bloom/v3 v3.7.0 github.com/coreos/go-systemd/v22 v22.5.0 + github.com/creack/pty v1.1.23 github.com/gokrazy/rsync v0.2.10 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 github.com/klauspost/compress v1.18.0 @@ -32,7 +32,6 @@ require ( require ( github.com/bits-and-blooms/bitset v1.24.0 // indirect - github.com/creack/pty v1.1.23 // indirect github.com/fsnotify/fsnotify v1.8.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect diff --git a/go.sum b/go.sum index 3b114736..7912aa54 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,3 @@ -bou.ke/monkey v1.0.2 h1:kWcnsrCNUatbxncxR/ThdYqbytgOIArtYWqcQLQzKLI= -bou.ke/monkey v1.0.2/go.mod h1:OqickVX3tNx6t33n1xvtTtu85YN5s6cKwVug+oHMaIA= github.com/bits-and-blooms/bitset v1.10.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= github.com/bits-and-blooms/bitset v1.24.0 h1:H4x4TuulnokZKvHLfzVRTHJfFfnHEeSYJizujEZvmAM= github.com/bits-and-blooms/bitset v1.24.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= diff --git a/transfer/save_checksum_state_test.go b/transfer/save_checksum_state_test.go index 71ef6244..0b348b6e 100644 --- a/transfer/save_checksum_state_test.go +++ b/transfer/save_checksum_state_test.go @@ -4,10 +4,8 @@ import ( "errors" "os" "path/filepath" - "reflect" "testing" - "bou.ke/monkey" "go.uber.org/zap" "go.uber.org/zap/zaptest/observer" ) @@ -22,22 +20,15 @@ func TestSaveChecksumState(t *testing.T) { }) t.Run("close warning", func(t *testing.T) { - var f *os.File - patchChmod := monkey.PatchInstanceMethod(reflect.TypeOf(f), "Chmod", func(*os.File, os.FileMode) error { - return errors.New("chmod fail") - }) - defer patchChmod.Unpatch() - patchClose := monkey.PatchInstanceMethod(reflect.TypeOf(f), "Close", func(*os.File) error { - return errors.New("close fail") - }) - defer patchClose.Unpatch() - core, observed := observer.New(zap.WarnLevel) logger := zap.New(core) - - filename := filepath.Join(t.TempDir(), "state") state := &ChecksumState{Checksums: make(map[uint64][]byte), Strategy: "sha256"} - if err := SaveChecksumState(filename, state, logger); err == nil { + ff := &fakeFile{ + chmodErr: errors.New("chmod fail"), + closeErr: errors.New("close fail"), + } + err := saveChecksumState("ignored", state, logger, func(string) (checksumFile, error) { return ff, nil }) + if err == nil { t.Fatalf("expected SaveChecksumState error") } logs := observed.FilterMessage("Failed to close checksum state file").All() @@ -46,3 +37,12 @@ func TestSaveChecksumState(t *testing.T) { } }) } + +type fakeFile struct { + chmodErr error + closeErr error +} + +func (f *fakeFile) Write(p []byte) (int, error) { return len(p), nil } +func (f *fakeFile) Chmod(os.FileMode) error { return f.chmodErr } +func (f *fakeFile) Close() error { return f.closeErr } diff --git a/transfer/transfer.go b/transfer/transfer.go index 599fa183..a0477644 100644 --- a/transfer/transfer.go +++ b/transfer/transfer.go @@ -82,13 +82,14 @@ func LoadChecksumState(filename string) (state *ChecksumState, err error) { return state, nil } -// SaveChecksumState persists block checksums. logger must be non-nil; use -// zap.NewNop() to disable logging. -// -//revive:disable-next-line:cognitive-complexity -func SaveChecksumState(filename string, state *ChecksumState, logger *zap.Logger) (err error) { - var file *os.File - file, err = os.Create(filename) +type checksumFile interface { + io.WriteCloser + Chmod(os.FileMode) error +} + +func saveChecksumState(filename string, state *ChecksumState, logger *zap.Logger, create func(string) (checksumFile, error)) (err error) { + var file checksumFile + file, err = create(filename) if err != nil { return fmt.Errorf("create checksum state: %w", err) } @@ -108,6 +109,14 @@ func SaveChecksumState(filename string, state *ChecksumState, logger *zap.Logger return nil } +// SaveChecksumState persists block checksums. logger must be non-nil; use +// zap.NewNop() to disable logging. +func SaveChecksumState(filename string, state *ChecksumState, logger *zap.Logger) error { + return saveChecksumState(filename, state, logger, func(name string) (checksumFile, error) { + return os.Create(name) + }) +} + // dumpChangesCore handles core transfer logic; logger must be non-nil. func (t *Transfer) dumpChangesCore(ctx context.Context, cfg *config.Config, snapshot, source string, out io.Writer, dedup DeduplicationStrategy, handshake string) (err error) { defer rootcmd.SyncLogger(t.Logger)