fix: harden copy pipeline and testing

AGENTS.md

@@ -0,0 +1,65 @@
+# Repository Guide for Agents
+
+## Scope
+
+This file applies to the entire repository.
+
+All source files, comments, tests, error messages, and documentation in this repository must be written in English.
+
+## Project overview
+
+`acp` is an advanced file copy tool and Go library. Its main features include concurrent copy workers, ordered traversal for linear devices, memory-mapped reads, multiple destinations, file metadata preservation, progress events, and JSON reports.
+
+The repository also contains `cmd/acp-rewrite`, which rewrites regular files through temporary copies, persists resumable state, preserves hardlink groups, and reports duplicate content by size and SHA256.
+
+## Main data flow
+
+The copy pipeline is:
+
+1. `index`: walk sources, validate regular files, collect metadata, and order jobs.
+2. `prepare`: open each source through a normal file or memory mapping.
+3. `copy`: stream each source to one or more destination writers and optionally calculate SHA256.
+4. `cleanup`: restore metadata and publish the final job status.
+5. `eventLoop`: fan out progress, job, and error events to registered handlers.
+
+Pipeline changes must preserve channel ownership, wait for every worker, propagate failures into reports, and avoid leaving source readers or destination files open.
+
+## Current maintenance context
+
+The current change set addresses these areas:
+
+- make memoized cache entries safe for concurrent use with `sync.Map` and per-key `sync.Once`;
+- support empty files without passing a nil reader into the copy pipeline;
+- serialize concrete report errors without unsafe interface pointer conversion;
+- retry failed hardlink relinking through persisted rewrite state;
+- select the longest matching mountpoint with directory-boundary checks;
+- use `path/filepath` for host file-system paths;
+- centralize platform-aware path ordering in `comparePath`;
+- add semantic, race, cross-platform, and CLI end-to-end coverage.
+
+The working tree may already contain staged or unstaged changes. Preserve them and do not reset, rewrite, or discard unrelated work.
+
+## Change guidelines
+
+- Keep changes narrow and follow the style of the surrounding package.
+- Use `path/filepath` for local file-system paths. Use `path` only for slash-delimited logical paths.
+- Keep path ordering platform-aware through the shared `comparePath` function.
+- Preserve the `Cache` guarantee that one key is initialized once while allowing different keys to initialize concurrently.
+- Treat a zero-length regular file as a valid copy job.
+- Keep report errors round-trippable, including messages that contain percent signs.
+- Persist retry state before considering a rewrite or hardlink operation complete.
+- Preserve build-tagged behavior in `syscall_*`, `mmap/*`, and `cmd/acp-rewrite/file_*` files.
+- Do not add concurrency unless it has explicit ownership, completion, and error semantics.
+- Add semantic regression tests for every behavior change.
+
+## Verification
+
+Follow [TESTING.md](TESTING.md). At minimum, run:
+
+```sh
+go test ./...
+go test -race ./...
+go vet ./...
+```
+
+Cross-build for Linux and Windows after changing path handling, system calls, memory mapping, or file metadata.

README.md

@@ -41,3 +41,7 @@ acp example -target target1 -target target2
 # do not copy, just get a dir index, write to `report.json`
 acp example -notarget -report report.json
 ```
+
+## Testing
+
+See [TESTING.md](TESTING.md) for unit, race, end-to-end, and cross-platform test instructions.

TESTING.md

@@ -0,0 +1,93 @@
+# Testing
+
+Run all commands from the repository root.
+
+## Prerequisites
+
+- Go 1.18 or newer.
+- Enough free space under the system temporary directory for test files and compiled binaries.
+- A host file system that supports the operations exercised by the selected tests.
+
+## Standard checks
+
+Run the complete test suite:
+
+```sh
+go test ./...
+```
+
+Run static analysis:
+
+```sh
+go vet ./...
+```
+
+Run the suite with the race detector:
+
+```sh
+go test -race ./...
+```
+
+## End-to-end test
+
+The end-to-end test builds the `acp` command, copies a directory tree through the CLI, and verifies:
+
+- regular and empty file contents;
+- nested destination paths;
+- successful job status and destinations;
+- JSON report decoding;
+- SHA256 generation for every copied file.
+
+Run only this test with:
+
+```sh
+go test -run '^TestACPCopyE2E$' -v .
+```
+
+The test is skipped when `go test` is run with `-short`.
+
+## Focused tests
+
+Run cache concurrency tests:
+
+```sh
+go test -race -run '^TestCache' -count=20 .
+```
+
+Run path and mountpoint tests:
+
+```sh
+go test -run '^(TestComparePath|TestSourceRoot|TestFindMountpoint)$' .
+```
+
+Run `acp-rewrite` tests:
+
+```sh
+go test ./cmd/acp-rewrite
+```
+
+## Cross-platform checks
+
+Build all packages for Linux and Windows:
+
+```sh
+GOOS=linux GOARCH=amd64 go build ./...
+GOOS=windows GOARCH=amd64 go build ./...
+```
+
+Compile the Windows-specific root package tests without running them:
+
+```sh
+GOOS=windows GOARCH=amd64 go test -c -o /tmp/acp-windows.test .
+```
+
+## Change checklist
+
+Before submitting a change:
+
+1. Run `gofmt` on modified Go files.
+2. Run `go test ./...`.
+3. Run `go test -race ./...` for concurrency, cache, pipeline, or event changes.
+4. Run `go vet ./...`.
+5. Cross-build when changing paths, system calls, memory mapping, or file metadata.
+6. Confirm that source files, comments, tests, and documentation contain only English text.

acp.go

@@ -53,9 +53,7 @@ func (c *Copyer) Wait() {
 
 func (c *Copyer) run(ctx context.Context) error {
 	defer c.running.Done()
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
+	defer close(c.eventCh)
 
 	go wrap(ctx, func() { c.eventLoop(ctx) })
 
@@ -112,17 +110,9 @@ func (c *Copyer) eventLoop(ctx context.Context) {
 			close(ch)
 		}
 	}()
-	for {
-		select {
-		case e, ok := <-c.eventCh:
-			if !ok {
-				return
-			}
-			for _, ch := range chans {
-				ch <- e
-			}
-		case <-ctx.Done():
-			return
+	for e := range c.eventCh {
+		for _, ch := range chans {
+			ch <- e
 		}
 	}
 }

cache.go

@@ -1,15 +1,22 @@
 package acp
 
+import "sync"
+
+type cacheEntry[V any] struct {
+	once  sync.Once
+	value V
+}
+
 func Cache[K comparable, V any](f func(in K) V) func(in K) V {
-	cache := make(map[K]V, 0)
+	var cache sync.Map
 	return func(in K) V {
-		cached, has := cache[in]
-		if has {
-			return cached
+		cached, has := cache.Load(in)
+		if !has {
+			cached, _ = cache.LoadOrStore(in, new(cacheEntry[V]))
 		}
 
-		out := f(in)
-		cache[in] = out
-		return out
+		entry := cached.(*cacheEntry[V])
+		entry.once.Do(func() { entry.value = f(in) })
+		return entry.value
 	}
 }

cache_test.go

@@ -0,0 +1,72 @@
+package acp
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestCacheConcurrentInitOnce(t *testing.T) {
+	var initialized int64
+	cached := Cache(func(in int) *int {
+		atomic.AddInt64(&initialized, 1)
+		out := new(int)
+		*out = in
+		return out
+	})
+
+	const workers = 64
+	start := make(chan struct{})
+	results := make(chan *int, workers)
+	var wg sync.WaitGroup
+	for idx := 0; idx < workers; idx++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			results <- cached(1)
+		}()
+	}
+
+	close(start)
+	wg.Wait()
+	close(results)
+
+	if got := atomic.LoadInt64(&initialized); got != 1 {
+		t.Fatalf("cache initialized %d times", got)
+	}
+
+	var first *int
+	for result := range results {
+		if first == nil {
+			first = result
+			continue
+		}
+		if result != first {
+			t.Fatalf("cache returned different values")
+		}
+	}
+}
+
+func TestCacheInitializerCanLoadAnotherKey(t *testing.T) {
+	var cached func(int) int
+	cached = Cache(func(in int) int {
+		if in == 0 {
+			return 0
+		}
+		return cached(in-1) + 1
+	})
+
+	done := make(chan int, 1)
+	go func() { done <- cached(2) }()
+
+	select {
+	case got := <-done:
+		if got != 2 {
+			t.Fatalf("cached value = %d", got)
+		}
+	case <-time.After(time.Second):
+		t.Fatalf("cache initializer deadlocked")
+	}
+}

cmd/acp-rewrite/main.go

@@ -212,7 +212,7 @@ func main() {
 			continue
 		}
 
-		if err := relink(entry); err != nil {
+		if err := relinkOrRetry(state, entry); err != nil {
 			logrus.Warnf("relink fail, path= '%s', err= %s", entry.Path, err)
 		}
 
@@ -429,6 +429,14 @@ func relink(entry rewriteEntry) error {
 	return nil
 }
 
+func relinkOrRetry(state *rewriteState, entry rewriteEntry) error {
+	if err := relink(entry); err != nil {
+		state.Pending = append(state.Pending, entry)
+		return err
+	}
+	return nil
+}
+
 func relinkOne(src, link string) error {
 	tmpPath, err := newTmpPath(link, randSource)
 	if err != nil {

cmd/acp-rewrite/main_test.go

@@ -182,3 +182,22 @@ func TestRelinkOnePreserveLinkAttrs(t *testing.T) {
 		t.Fatalf("link content = %q", string(data))
 	}
 }
+
+func TestRelinkOrRetry(t *testing.T) {
+	dir := t.TempDir()
+	entry := rewriteEntry{
+		Path:  filepath.Join(dir, "missing-src"),
+		Links: []string{filepath.Join(dir, "link")},
+	}
+	state := new(rewriteState)
+
+	if err := relinkOrRetry(state, entry); err == nil {
+		t.Fatalf("expected relink error")
+	}
+	if len(state.Pending) != 1 {
+		t.Fatalf("pending entries = %d", len(state.Pending))
+	}
+	if state.Pending[0].Path != entry.Path {
+		t.Fatalf("pending path = %q", state.Pending[0].Path)
+	}
+}

copy.go

@@ -7,7 +7,7 @@ import (
 	"hash"
 	"io"
 	"os"
-	"path"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -30,18 +30,22 @@ func (c *Copyer) copy(ctx context.Context, prepared <-chan *writeJob) <-chan *ba
 
 	var copying sync.WaitGroup
 	done := make(chan struct{})
+	var reporting sync.WaitGroup
 	defer func() {
 		go wrap(ctx, func() {
-			defer close(done)
-			defer close(ch)
-
 			copying.Wait()
+			close(done)
+			reporting.Wait()
+			close(ch)
 		})
 	}()
 
 	cntr := new(counter)
+	reporting.Add(1)
 	go wrap(ctx, func() {
+		defer reporting.Done()
 		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ticker.C:
@@ -121,7 +125,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c
 			continue
 		}
 
-		if err := mappingError(os.MkdirAll(path.Dir(target), os.ModePerm)); err != nil {
+		if err := mappingError(os.MkdirAll(filepath.Dir(target), os.ModePerm)); err != nil {
 			if checkErrorAbort(err) {
 				noSpaceDevices.Add(dev)
 			}
@@ -139,7 +143,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c
 			job.fail(target, fmt.Errorf("open dst file fail, %w", err))
 			continue
 		}
-		if !job.copyer.toDevice.linear {
+		if !job.copyer.toDevice.linear && job.size > 0 {
 			if err := truncate(file, job.size); err != nil {
 				job.fail(target, fmt.Errorf("truncate dst file fail, %w", err))
 				continue

copy_test.go

@@ -0,0 +1,75 @@
+package acp
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestCopyEmptyFile(t *testing.T) {
+	tests := []struct {
+		name string
+		opts []Option
+	}{
+		{name: "mmap"},
+		{name: "linear", opts: []Option{SetFromDevice(LinearDevice(true))}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dir := t.TempDir()
+			src := filepath.Join(dir, "src")
+			dst := filepath.Join(dir, "dst")
+			if err := os.WriteFile(src, nil, 0o644); err != nil {
+				t.Fatalf("write src: %v", err)
+			}
+
+			handler, getter := NewReportGetter()
+			opts := append([]Option{
+				AccurateJob(src, []string{dst}),
+				Overwrite(true),
+				WithEventHandler(handler),
+			}, tt.opts...)
+			copyer, err := New(context.Background(), opts...)
+			if err != nil {
+				t.Fatalf("new copyer: %v", err)
+			}
+
+			done := make(chan struct{})
+			go func() {
+				copyer.Wait()
+				close(done)
+			}()
+			select {
+			case <-done:
+			case <-time.After(5 * time.Second):
+				t.Fatalf("copy empty file timed out")
+			}
+
+			info, err := os.Stat(dst)
+			if err != nil {
+				t.Fatalf("stat dst: %v", err)
+			}
+			if info.Size() != 0 {
+				t.Fatalf("dst size = %d", info.Size())
+			}
+
+			report := getter()
+			if len(report.Errors) != 0 {
+				t.Fatalf("report errors = %v", report.Errors)
+			}
+			if len(report.Jobs) != 1 {
+				t.Fatalf("report jobs = %d", len(report.Jobs))
+			}
+			job := report.Jobs[0]
+			if job.Status != JobStatusFinished {
+				t.Fatalf("job status = %q", job.Status)
+			}
+			if len(job.SuccessTargets) != 1 || job.SuccessTargets[0] != dst {
+				t.Fatalf("success targets = %v", job.SuccessTargets)
+			}
+		})
+	}
+}

e2e_test.go

@@ -0,0 +1,119 @@
+package acp_test
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/samuelncui/acp"
+)
+
+func TestACPCopyE2E(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping end-to-end test in short mode")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	repoRoot, err := os.Getwd()
+	if err != nil {
+		t.Fatalf("get working directory: %v", err)
+	}
+	tempDir := t.TempDir()
+	binary := filepath.Join(tempDir, "acp")
+	if runtime.GOOS == "windows" {
+		binary += ".exe"
+	}
+
+	run := func(name string, args ...string) {
+		t.Helper()
+		cmd := exec.CommandContext(ctx, name, args...)
+		cmd.Dir = repoRoot
+		if output, err := cmd.CombinedOutput(); err != nil {
+			t.Fatalf("run %s: %v\n%s", name, err, output)
+		}
+	}
+	run("go", "build", "-o", binary, "./cmd/acp")
+
+	source := filepath.Join(tempDir, "source")
+	target := filepath.Join(tempDir, "target")
+	reportPath := filepath.Join(tempDir, "report.json")
+	if err := os.MkdirAll(target, 0o755); err != nil {
+		t.Fatalf("create target directory: %v", err)
+	}
+
+	files := map[string][]byte{
+		"plain.txt":                     []byte("plain text\n"),
+		"empty.txt":                     nil,
+		filepath.Join("nested", "data"): {0, 1, 2, 3, 4},
+	}
+	for relativePath, data := range files {
+		path := filepath.Join(source, relativePath)
+		if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+			t.Fatalf("create source directory: %v", err)
+		}
+		if err := os.WriteFile(path, data, 0o644); err != nil {
+			t.Fatalf("write source file %q: %v", relativePath, err)
+		}
+	}
+
+	run(binary, "-p=false", "-report", reportPath, source, target)
+
+	copyRoot := filepath.Join(target, filepath.Base(source))
+	for relativePath, want := range files {
+		got, err := os.ReadFile(filepath.Join(copyRoot, relativePath))
+		if err != nil {
+			t.Fatalf("read copied file %q: %v", relativePath, err)
+		}
+		if !bytes.Equal(got, want) {
+			t.Fatalf("copied file %q = %v, want %v", relativePath, got, want)
+		}
+	}
+
+	reportData, err := os.ReadFile(reportPath)
+	if err != nil {
+		t.Fatalf("read report: %v", err)
+	}
+	var report acp.Report
+	if err := json.Unmarshal(reportData, &report); err != nil {
+		t.Fatalf("decode report: %v", err)
+	}
+	if len(report.Errors) != 0 {
+		t.Fatalf("report errors = %v", report.Errors)
+	}
+	if len(report.Jobs) != len(files) {
+		t.Fatalf("report jobs = %d, want %d", len(report.Jobs), len(files))
+	}
+
+	jobs := make(map[string]*acp.Job, len(report.Jobs))
+	for _, job := range report.Jobs {
+		jobs[job.FullPath] = job
+	}
+	for relativePath := range files {
+		sourcePath := filepath.Join(source, relativePath)
+		job, ok := jobs[sourcePath]
+		if !ok {
+			t.Fatalf("report job not found for %q", sourcePath)
+		}
+		if job.Status != acp.JobStatusFinished {
+			t.Fatalf("job %q status = %q", sourcePath, job.Status)
+		}
+		if len(job.FailTargets) != 0 {
+			t.Fatalf("job %q failures = %v", sourcePath, job.FailTargets)
+		}
+		wantTarget := filepath.Join(copyRoot, relativePath)
+		if len(job.SuccessTargets) != 1 || job.SuccessTargets[0] != wantTarget {
+			t.Fatalf("job %q success targets = %v", sourcePath, job.SuccessTargets)
+		}
+		if job.SHA256 == "" {
+			t.Fatalf("job %q has no SHA256", sourcePath)
+		}
+	}
+}

error.go

@@ -2,6 +2,7 @@ package acp
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 )
 
@@ -37,6 +38,6 @@ func (e *Error) UnmarshalJSON(buf []byte) error {
 		return err
 	}
 
-	e.Src, e.Dst, e.Err = m.Src, m.Dst, fmt.Errorf(m.Err)
+	e.Src, e.Dst, e.Err = m.Src, m.Dst, errors.New(m.Err)
 	return nil
 }

fs.go

@@ -24,12 +24,7 @@ func getMountpointCache() (func(string) string, error) {
 			continue
 		}
 
-		mp := mount.Mountpoint
-		if !strings.HasSuffix(mp, "/") {
-			mp = mp + "/"
-		}
-
-		mountPoints.Add(mp)
+		mountPoints.Add(filepath.Clean(mount.Mountpoint))
 	}
 
 	mps := mountPoints.ToSlice()
@@ -39,11 +34,25 @@ func getMountpointCache() (func(string) string, error) {
 			panic(fmt.Errorf("get abs from file path failed, path= '%s', %w", path, err))
 		}
 
-		for _, mp := range mps {
-			if strings.HasPrefix(path, mp) {
-				return mp
-			}
-		}
-		return ""
+		return findMountpoint(path, mps)
 	}), nil
 }
+
+func findMountpoint(path string, mountPoints []string) string {
+	matched := ""
+	for _, mountPoint := range mountPoints {
+		if path != mountPoint {
+			prefix := mountPoint
+			if !strings.HasSuffix(prefix, string(filepath.Separator)) {
+				prefix += string(filepath.Separator)
+			}
+			if !strings.HasPrefix(path, prefix) {
+				continue
+			}
+		}
+		if len(mountPoint) > len(matched) {
+			matched = mountPoint
+		}
+	}
+	return matched
+}

fs_test.go

@@ -1,6 +1,9 @@
 package acp
 
-import "testing"
+import (
+	"path/filepath"
+	"testing"
+)
 
 func TestFS(t *testing.T) {
 	mpCache, err := getMountpointCache()
@@ -10,3 +13,29 @@ func TestFS(t *testing.T) {
 
 	t.Log("mp cahce", mpCache("/Users/cuijingning/go/src/github.com/samuelncui/acp"))
 }
+
+func TestFindMountpoint(t *testing.T) {
+	root := string(filepath.Separator)
+	mount := filepath.Join(root, "mnt")
+	nested := filepath.Join(mount, "archive")
+	mountPoints := []string{root, nested, mount}
+
+	tests := []struct {
+		name string
+		path string
+		want string
+	}{
+		{name: "nested child", path: filepath.Join(nested, "file"), want: nested},
+		{name: "exact mountpoint", path: nested, want: nested},
+		{name: "parent child", path: filepath.Join(mount, "file"), want: mount},
+		{name: "directory boundary", path: filepath.Join(root, "mnt2", "file"), want: root},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := findMountpoint(tt.path, mountPoints); got != tt.want {
+				t.Fatalf("findMountpoint(%q) = %q, want %q", tt.path, got, tt.want)
+			}
+		})
+	}
+}

index.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"os"
 	"sort"
-	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 )
@@ -42,11 +42,18 @@ func (c *Copyer) index(ctx context.Context) (<-chan *baseJob, error) {
 
 func (c *Copyer) walk(ctx context.Context) ([]*baseJob, error) {
 	done := make(chan struct{})
-	defer close(done)
+	var reporting sync.WaitGroup
+	reporting.Add(1)
+	defer func() {
+		close(done)
+		reporting.Wait()
+	}()
 
 	cntr := new(counter)
 	go wrap(ctx, func() {
+		defer reporting.Done()
 		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ticker.C:
@@ -173,8 +180,7 @@ func (c *Copyer) walk(ctx context.Context) ([]*baseJob, error) {
 
 func (c *Copyer) joinJobs(jobs []*baseJob) ([]*baseJob, error) {
 	sort.Slice(jobs, func(i int, j int) bool {
-		si, sj := strings.ReplaceAll(jobs[i].src.path, "/", "\x00"), strings.ReplaceAll(jobs[j].src.path, "/", "\x00")
-		return si < sj
+		return comparePath(jobs[i].src.path, jobs[j].src.path) < 0
 	})
 
 	var last *baseJob

job.go

@@ -4,7 +4,6 @@ import (
 	"encoding/hex"
 	"io"
 	"io/fs"
-	"path"
 	"sync"
 	"time"
 )
@@ -93,7 +92,7 @@ func (j *baseJob) fail(path string, err error) {
 
 func (j *baseJob) report() *Job {
 	return &Job{
-		FullPath: path.Join(j.src.base, j.src.path),
+		FullPath: j.path,
 		Base:     j.src.base,
 		Path:     j.src.path,
 

opt.go

@@ -2,7 +2,8 @@ package acp
 
 import (
 	"os"
-	"path"
+	"path/filepath"
+	"strings"
 
 	"github.com/sirupsen/logrus"
 )
@@ -13,15 +14,21 @@ type source struct {
 }
 
 func (s *source) src() string {
-	return path.Join(s.base, s.path)
+	return filepath.Join(s.base, s.path)
 }
 
 func (s *source) dst(dst string) string {
-	return path.Join(dst, s.path)
+	return filepath.Join(dst, s.path)
 }
 
 func (s *source) append(next string) *source {
-	return &source{base: s.base, path: path.Join(s.path, next)}
+	return &source{base: s.base, path: filepath.Join(s.path, next)}
+}
+
+func comparePath(a, b string) int {
+	a = strings.ReplaceAll(filepath.ToSlash(a), "/", "\x00")
+	b = strings.ReplaceAll(filepath.ToSlash(b), "/", "\x00")
+	return strings.Compare(a, b)
 }
 
 type option struct {
@@ -75,7 +82,7 @@ type accurateJob struct {
 
 func AccurateJob(src string, dsts []string) Option {
 	return func(o *option) *option {
-		o.accurateJobs = append(o.accurateJobs, &accurateJob{src: path.Clean(src), dsts: dsts})
+		o.accurateJobs = append(o.accurateJobs, &accurateJob{src: filepath.Clean(src), dsts: dsts})
 		return o
 	}
 }

opt_test.go

@@ -0,0 +1,53 @@
+package acp
+
+import (
+	"path/filepath"
+	"sort"
+	"testing"
+)
+
+func TestSourceRoot(t *testing.T) {
+	root := string(filepath.Separator)
+	job := Source(root)(new(wildcardJob))
+	if len(job.src) != 1 {
+		t.Fatalf("sources = %d", len(job.src))
+	}
+
+	src := job.src[0]
+	if got := src.src(); got != root {
+		t.Fatalf("source root = %q, want %q", got, root)
+	}
+
+	target := filepath.Join(root, "target")
+	child := src.append("file")
+	if got := child.src(); got != filepath.Join(root, "file") {
+		t.Fatalf("child source = %q", got)
+	}
+	if got := child.dst(target); got != filepath.Join(target, "file") {
+		t.Fatalf("child target = %q", got)
+	}
+}
+
+func TestComparePath(t *testing.T) {
+	paths := []string{
+		"b",
+		"a-b",
+		filepath.Join("a", "b", "c"),
+		"a",
+		filepath.Join("a", "b"),
+	}
+	sort.Slice(paths, func(i, j int) bool { return comparePath(paths[i], paths[j]) < 0 })
+
+	want := []string{
+		"a",
+		filepath.Join("a", "b"),
+		filepath.Join("a", "b", "c"),
+		"a-b",
+		"b",
+	}
+	for idx := range want {
+		if paths[idx] != want[idx] {
+			t.Fatalf("paths[%d] = %q, want %q", idx, paths[idx], want[idx])
+		}
+	}
+}

opt_wildcard.go

@@ -3,7 +3,7 @@ package acp
 import (
 	"fmt"
 	"os"
-	"path"
+	"path/filepath"
 	"sort"
 	"strings"
 )
@@ -20,9 +20,7 @@ func (job *wildcardJob) check() error {
 		if p == "" {
 			continue
 		}
-		if p[len(p)-1] != '/' {
-			p = p + "/"
-		}
+		p = filepath.Clean(p)
 
 		dstStat, err := os.Stat(p)
 		if err != nil {
@@ -40,8 +38,7 @@ func (job *wildcardJob) check() error {
 		return fmt.Errorf("source path not found")
 	}
 	sort.Slice(job.src, func(i, j int) bool {
-		si, sj := strings.ReplaceAll(job.src[i].path, "/", "\x00"), strings.ReplaceAll(job.src[j].path, "/", "\x00")
-		return si < sj
+		return comparePath(job.src[i].path, job.src[j].path) < 0
 	})
 	for _, s := range job.src {
 		src := s.src()
@@ -74,12 +71,8 @@ type WildcardJobOption func(*wildcardJob) *wildcardJob
 func Source(paths ...string) WildcardJobOption {
 	return func(j *wildcardJob) *wildcardJob {
 		for _, p := range paths {
-			p = path.Clean(p)
-			if p[len(p)-1] == '/' {
-				p = p[:len(p)-1]
-			}
-
-			base, name := path.Split(p)
+			p = filepath.Clean(p)
+			base, name := filepath.Split(p)
 			j.src = append(j.src, &source{base: base, path: name})
 		}
 		return j
@@ -99,7 +92,7 @@ func SourceWithPath(base string, paths ...string) WildcardJobOption {
 func AccurateSource(base string, paths ...[]string) WildcardJobOption {
 	return func(j *wildcardJob) *wildcardJob {
 		for _, p := range paths {
-			j.src = append(j.src, &source{base: base, path: path.Join(p...)})
+			j.src = append(j.src, &source{base: base, path: filepath.Join(p...)})
 		}
 		return j
 	}

opt_windows_test.go

@@ -0,0 +1,35 @@
+//go:build windows
+
+package acp
+
+import "testing"
+
+func TestSourceWindowsPath(t *testing.T) {
+	job := Source(`C:\src\file.txt`)(new(wildcardJob))
+	if len(job.src) != 1 {
+		t.Fatalf("sources = %d", len(job.src))
+	}
+
+	src := job.src[0]
+	if got := src.src(); got != `C:\src\file.txt` {
+		t.Fatalf("source = %q", got)
+	}
+	if got := src.dst(`D:\target`); got != `D:\target\file.txt` {
+		t.Fatalf("target = %q", got)
+	}
+}
+
+func TestSourceWindowsRoot(t *testing.T) {
+	job := Source(`C:\`)(new(wildcardJob))
+	if len(job.src) != 1 {
+		t.Fatalf("sources = %d", len(job.src))
+	}
+
+	src := job.src[0]
+	if got := src.src(); got != `C:\` {
+		t.Fatalf("source root = %q", got)
+	}
+	if got := src.append("file").src(); got != `C:\file` {
+		t.Fatalf("child source = %q", got)
+	}
+}

prepare.go

@@ -1,6 +1,7 @@
 package acp
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
@@ -50,11 +51,9 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 
 							fileInfo, err := file.Stat()
 							if err != nil {
+								_ = file.Close()
 								return nil, 0, fmt.Errorf("get src file stat fail, %w", err)
 							}
-							if fileInfo.Size() == 0 {
-								return nil, 0, fmt.Errorf("get src file, size is zero")
-							}
 
 							return file, fileInfo.Size(), nil
 						}
@@ -64,13 +63,19 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 							return nil, 0, fmt.Errorf("open src file by mmap fail, %w", err)
 						}
 						if readerAt.Len() == 0 {
-							return nil, 0, fmt.Errorf("get src file by mmap, size is zero")
+							if err := readerAt.Close(); err != nil {
+								return nil, 0, fmt.Errorf("close empty src file by mmap fail, %w", err)
+							}
+							return io.NopCloser(bytes.NewReader(nil)), 0, nil
 						}
 
 						return mmap.NewReader(readerAt), int64(readerAt.Len()), nil
 					}(job.path)
 					if err != nil {
 						c.reportError(job.path, "", err)
+						job.fail("", err)
+						job.setStatus(jobStatusFinished)
+						continue
 					}
 
 					wj := newWriteJob(job, file, size, c.fromDevice.linear)

report.go

@@ -1,7 +1,7 @@
 package acp
 
 import (
-	"fmt"
+	"errors"
 	"sync"
 	"unsafe"
 
@@ -89,7 +89,7 @@ func (*errValCoder) Encode(ptr unsafe.Pointer, stream *jsoniter.Stream) {
 
 func (*errValCoder) Decode(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 	val := (*error)(ptr)
-	*val = fmt.Errorf(iter.ReadString())
+	*val = errors.New(iter.ReadString())
 }
 
 var (
@@ -101,14 +101,14 @@ type reportJSONExtension struct {
 }
 
 func (*reportJSONExtension) CreateDecoder(typ reflect2.Type) jsoniter.ValDecoder {
-	if typ.Implements(errorType2) {
+	if typ == errorType2 {
 		return &errValCoder{}
 	}
 	return nil
 }
 
 func (*reportJSONExtension) CreateEncoder(typ reflect2.Type) jsoniter.ValEncoder {
-	if typ.Implements(errorType2) {
+	if typ == errorType2 {
 		return &errValCoder{}
 	}
 	return nil

report_test.go

@@ -1,6 +1,7 @@
 package acp
 
 import (
+	"errors"
 	"syscall"
 	"testing"
 
@@ -22,3 +23,29 @@ func TestErrorJSONMarshal(t *testing.T) {
 	buf, _ := reportJSON.Marshal(m)
 	logrus.Infof("get json %s", buf)
 }
+
+func TestReportJSONErrorRoundTrip(t *testing.T) {
+	const message = "copy 100% failed"
+	want := &Report{
+		Jobs: []*Job{{
+			FailTargets: map[string]error{"dst": errors.New(message)},
+		}},
+		Errors: []*Error{{Src: "src", Dst: "dst", Err: errors.New(message)}},
+	}
+
+	buf, err := reportJSON.Marshal(want)
+	if err != nil {
+		t.Fatalf("marshal report: %v", err)
+	}
+
+	var got Report
+	if err := reportJSON.Unmarshal(buf, &got); err != nil {
+		t.Fatalf("unmarshal report: %v", err)
+	}
+	if len(got.Errors) != 1 || got.Errors[0].Err.Error() != message {
+		t.Fatalf("report errors = %+v", got.Errors)
+	}
+	if len(got.Jobs) != 1 || got.Jobs[0].FailTargets["dst"].Error() != message {
+		t.Fatalf("fail targets = %+v", got.Jobs)
+	}
+}