diff --git a/README.md b/README.md index 0a9ac08..b3d822b 100644 --- a/README.md +++ b/README.md @@ -57,8 +57,13 @@ with admitted files and their target counts, rather than with Read permits alone Random ordinary outputs complete metadata, Sync and Close before rename. Linear outputs omit per-file Sync and Close independently after data writing; metadata/cache work on their owned -temporary paths follows closure. Distinct paths may publish concurrently; linear outputs to the -same resolved path preserve publication order, including equivalent relative and absolute names. +temporary paths follows closure. Linear sources, linear targets and single-writer targets preserve +publication order for the same resolved path, including equivalent relative and absolute names. +Their completion keys also group case-folded, canonically equivalent Unicode names conservatively, +so case or normalization aliases keep the same winner. On a case-sensitive filesystem this may +order otherwise distinct outputs; their actual names and contents remain distinct. This key is +not a filesystem identity test for every vendor-specific naming rule. Other keys publish independently; +the copy data flow does not wait for these ordinary-file completion dependencies. Results keep the caller's original target names, and the exclusive-rename fallback remains in force. A late linear Close or metadata error stops future admission once observed; it cannot retract already started writes. Results retain each started item's actual outcome, and completion waits diff --git a/TESTING.md b/TESTING.md index 6945bd7..f7c06f8 100644 --- a/TESTING.md +++ b/TESTING.md @@ -76,6 +76,7 @@ Use this inventory to select or update coverage; it is not a record of tests run | Input-method evaluation, reused source facts, target resolution and alias/overwrite validation | `item_test.go`, `stream_test.go`, `fs_test.go`, `target_test.go` | Public stream and shell copies in `library_e2e_test.go` | | Source ownership, one content pass/hash, read modes, empty files and multiple targets | `copy_test.go`, `item_test.go`, `fs_test.go` | Library and command E2E | | Staged replacement, failure preservation, final-path outcomes and owned temporary cleanup | `staged_target_test.go`, `target_test.go`, `internal/fileio/output_test.go`, `internal/fileio/temp_test.go` | Library/command E2E and rewrite failure/resume tests | +| Ordered completion of native path aliases without blocking independent outputs | `publication_serial_test.go`, `pipeline_liveness_test.go` | Native case/decomposition and relative/absolute copies in `target_identity_test.go` | | Metadata, hash policies, between-run cache staleness, cache warnings and descriptor use | `signature_test.go`, `signature_xattr_copy_test.go`, `cleanup_test.go` | Real cache copies in those tests; metadata and report hashes in library/command E2E | | Results, batching, cancellation, backpressure, linear ordering and event delivery | `item_test.go`, `stream_test.go`, `copy_test.go`, `opt_test.go` | Command interrupt, report and progress tests in `e2e_test.go` | | Panic handling, channel completion and resource release | `recover_test.go`, `copy_test.go`, `stream_test.go` | Caller callback/handler panic tests driving a real stream | @@ -132,6 +133,10 @@ across the fourteen maintained benchmark cases. Go calibrates iterations; do not counts or repeated whole-suite samples. Compare time, bytes/op and allocations/op, recording all raw values and percentage deltas. +For changes to completion ordering, `go test -run '^$' -bench '^BenchmarkPublicationOrder$' +-benchmem -count=1 .` isolates per-file key/dependency overhead for ASCII and Unicode names. +It complements the copy workloads; it does not measure filesystem or Tape throughput. + For rewrite persistence changes, also run `GOMAXPROCS=4 make bench-rewrite` once per source. Its 100-entry and 1,000-entry fixtures measure one state/report checkpoint pair using the same harness on both sources. Full snapshots still serialize the whole queue and accumulated report; diff --git a/go.mod b/go.mod index d9bdb46..17d2a94 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/schollz/progressbar/v3 v3.13.1 github.com/sirupsen/logrus v1.9.3 golang.org/x/sys v0.48.0 + golang.org/x/text v0.39.0 ) require ( diff --git a/go.sum b/go.sum index c48d9d8..cddddc6 100644 --- a/go.sum +++ b/go.sum @@ -41,6 +41,8 @@ golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U= golang.org/x/term v0.12.0 h1:/ZfYdc3zq+q02Rv9vGqTeSItdzZTSNDmfTi0mBAuidU= golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU= +golang.org/x/text v0.39.0 h1:UbZz4pLOvn600D6Oh6GGEI6VAmndrEBLv8/6BEXzyus= +golang.org/x/text v0.39.0/go.mod h1:3UwRclnC2g0TU9x8PZiyfOajCd1zaUNHF9cvqcQZ+ZM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/pipeline_b_test.go b/pipeline_b_test.go index d5efdb2..1fd0b7c 100644 --- a/pipeline_b_test.go +++ b/pipeline_b_test.go @@ -28,28 +28,49 @@ func onceRelease(ch chan struct{}) func() { return func() { once.Do(func() { close(ch) }) } } +func cleanupPipeline(t *testing.T, c *StreamCopyer, releases ...func()) { + t.Helper() + + // Release fault-injection gates and join workers before model or global-hook cleanup. + t.Cleanup(func() { + for _, release := range releases { + release() + } + if err := c.Wait(); err != nil { + t.Errorf("pipeline teardown failed: %v", err) + } + }) +} + func TestPipelineOverlapsHashWriteAndLinearClose(t *testing.T) { - // Hold hashing and the first Close independently while both ordered data writes progress. + // Model hooks see resolved output paths; results retain the caller's original names. root := t.TempDir() + resolved, err := filepath.EvalSymlinks(root) + if err != nil { + t.Fatal(err) + } content := bytes.Repeat([]byte("parallel content"), 64<<10) source := writeSourceFile(t, root, "source", content) a, b := filepath.Join(root, "a"), filepath.Join(root, "b") + resolvedA, resolvedB := filepath.Join(resolved, "a"), filepath.Join(resolved, "b") + + // Hold hashing and the first Close independently while both ordered data writes progress. hashStarted, _, releaseHash := holdContentHash(t) - t.Cleanup(releaseHash) closeStarted, secondWrite, closeGate := make(chan struct{}), make(chan struct{}), make(chan struct{}) releaseClose := onceRelease(closeGate) - t.Cleanup(releaseClose) var closeOnce, writeOnce sync.Once model := newLTFSModel(ltfsModelConfig{capture: true, entered: func(op, path string) { - if op == "close" && path == a { + if op == "close" && path == resolvedA { closeOnce.Do(func() { close(closeStarted) }) <-closeGate } - if op == "write" && path == b { + if op == "write" && path == resolvedB { writeOnce.Do(func() { close(secondWrite) }) } }}) - defer model.shutdown() + t.Cleanup(model.shutdown) + + // Keep fixture cleanup behind every accepted transfer, including an early test failure. results := make(chan Result, 2) c, err := NewStream(context.Background(), func(batch []Result) error { for _, r := range batch { @@ -60,6 +81,7 @@ func TestPipelineOverlapsHashWriteAndLinearClose(t *testing.T) { if err != nil { t.Fatal(err) } + cleanupPipeline(t, c, releaseClose, releaseHash) c.filesystem = ltfsFilesystem{transferFilesystem: c.fs(), model: model} if err := c.Submit(&SimpleJob{Path: source, Dsts: []string{a}}, &SimpleJob{Path: source, Dsts: []string{b}}); err != nil { t.Fatal(err) @@ -94,9 +116,9 @@ func TestPipelineOverlapsHashWriteAndLinearClose(t *testing.T) { t.Fatalf("result: %+v", r) } } - for _, path := range []string{a, b} { + for _, path := range []string{resolvedA, resolvedB} { out := model.outputs[path] - if !out.committed || !bytes.Equal(out.data, content) { + if out == nil || !out.committed || !bytes.Equal(out.data, content) { t.Fatalf("output %q is incomplete", path) } } @@ -150,7 +172,6 @@ func TestPipelineLinearReadAdvancesBeforeSourceClose(t *testing.T) { b := writeSourceFile(t, root, "b", []byte("bbbb")) gate, closing, reading := make(chan struct{}), make(chan struct{}), make(chan struct{}) release := onceRelease(gate) - t.Cleanup(release) var closeOnce, readOnce sync.Once m := newLTFSModel(ltfsModelConfig{entered: func(op, path string) { if op == "source-close" && path == a { @@ -161,18 +182,23 @@ func TestPipelineLinearReadAdvancesBeforeSourceClose(t *testing.T) { readOnce.Do(func() { close(reading) }) } }}) - defer m.shutdown() + t.Cleanup(m.shutdown) + + // Join source closure before shutting down the model, even when an observation fails. var seen atomic.Int64 c, err := NewStream(context.Background(), func(r []Result) error { seen.Add(int64(len(r))); return nil }, WithHashPolicy(HashRead), SetFromDevice(LinearDevice(true))) if err != nil { t.Fatal(err) } + cleanupPipeline(t, c, release) c.filesystem = ltfsFilesystem{transferFilesystem: c.fs(), model: m, readPositions: map[string]int64{a: 0, b: 4}} if err := c.Submit(&SimpleJob{Path: a}, &SimpleJob{Path: b}); err != nil { t.Fatal(err) } done := make(chan struct{}) go func() { _ = c.Close(); close(done) }() + + // The next source can read while the preceding source still owns its Close. awaitStage(t, closing, "first source Close") awaitStage(t, reading, "second ordered Read") release() @@ -189,26 +215,38 @@ func TestPipelineLinearReadAdvancesBeforeSourceClose(t *testing.T) { } func TestPipelineLateCloseNoSpaceKeepsAlreadyStartedOutcome(t *testing.T) { - // The first late Close fails only after the second Write has entered; the third must stay unstarted. + // Model hooks see resolved output paths; results retain the caller's original names. root := t.TempDir() + resolved, err := filepath.EvalSymlinks(root) + if err != nil { + t.Fatal(err) + } src := writeSourceFile(t, root, "source", []byte("late close")) a, b, d := filepath.Join(root, "a"), filepath.Join(root, "b"), filepath.Join(root, "d") + resolvedA, resolvedB, resolvedD := filepath.Join(resolved, "a"), filepath.Join(resolved, "b"), filepath.Join(resolved, "d") + + // The first late Close fails only after the second Write has entered; the third must stay unstarted. secondStarted, secondGate := make(chan struct{}), make(chan struct{}) releaseSecond := onceRelease(secondGate) - t.Cleanup(releaseSecond) m := newLTFSModel(ltfsModelConfig{before: func(op, path string) { - if op == "write" && path == b { + if op == "write" && path == resolvedB { close(secondStarted) <-secondGate } }, closeError: func(path string) error { - if path == a { - <-secondStarted + if path == resolvedA { + // Teardown also releases this wait if the second writer never reaches the hook. + select { + case <-secondStarted: + case <-secondGate: + } return syscall.ENOSPC } return nil }}) - defer m.shutdown() + t.Cleanup(m.shutdown) + + // Retain all outcomes while cleanup can release a blocked writer after an assertion fails. results := make(chan Result, 3) c, err := NewStream(context.Background(), func(batch []Result) error { for _, r := range batch { @@ -219,13 +257,22 @@ func TestPipelineLateCloseNoSpaceKeepsAlreadyStartedOutcome(t *testing.T) { if err != nil { t.Fatal(err) } + cleanupPipeline(t, c, releaseSecond) c.filesystem = ltfsFilesystem{transferFilesystem: c.fs(), model: m} if err := c.Submit(&SimpleJob{Path: src, Dsts: []string{a}}, &SimpleJob{Path: src, Dsts: []string{b}}, &SimpleJob{Path: src, Dsts: []string{d}}); err != nil { t.Fatal(err) } done := make(chan struct{}) go func() { _ = c.Close(); close(done) }() - first := <-results + + // Observe the authoritative failure before allowing the already-started writer to finish. + awaitStage(t, secondStarted, "second Write before first Close failure") + var first Result + select { + case first = <-results: + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for the late Close failure") + } if len(first.Targets) != 1 || first.Targets[0].Path != a || !errors.Is(first.Targets[0].Err, syscall.ENOSPC) || !errors.Is(first.Targets[0].Err, ErrTargetNoSpace) { t.Fatalf("first result: %+v", first) } @@ -244,7 +291,9 @@ func TestPipelineLateCloseNoSpaceKeepsAlreadyStartedOutcome(t *testing.T) { if outcomes[b].Err != nil || outcomes[b].Targets[0].Err != nil || !errors.Is(outcomes[d].Err, ErrTargetNoSpace) { t.Fatalf("outcomes: %+v", outcomes) } - if m.outputs[d] != nil || !m.outputs[a].discarded || m.outputs[a].committed || !m.outputs[b].committed { + firstOutput, secondOutput := m.outputs[resolvedA], m.outputs[resolvedB] + if m.outputs[resolvedD] != nil || firstOutput == nil || secondOutput == nil || + !firstOutput.discarded || firstOutput.committed || !secondOutput.committed { t.Fatal("late Close publication/admission mismatch") } if err := m.validate(); err != nil { @@ -258,7 +307,6 @@ func TestPipelineRandomTargetsWriteConcurrently(t *testing.T) { src := writeSourceFile(t, root, "source", []byte("fanout")) gate := make(chan struct{}) release := onceRelease(gate) - t.Cleanup(release) entered := make(chan struct{}, 2) m := newLTFSModel(ltfsModelConfig{capture: true, before: func(op, path string) { if op == "write" { @@ -266,17 +314,22 @@ func TestPipelineRandomTargetsWriteConcurrently(t *testing.T) { <-gate } }}) - defer m.shutdown() + t.Cleanup(m.shutdown) + + // Release both consumers and join the run before model cleanup on every test exit. c, err := NewStream(context.Background(), func([]Result) error { return nil }, WithHashPolicy(HashRead), SetToDevice(DeviceThreads(1))) if err != nil { t.Fatal(err) } + cleanupPipeline(t, c, release) c.filesystem = ltfsFilesystem{transferFilesystem: c.fs(), model: m} if err := c.Submit(&SimpleJob{Path: src, Dsts: []string{filepath.Join(root, "a"), filepath.Join(root, "b")}}); err != nil { t.Fatal(err) } done := make(chan struct{}) go func() { _ = c.Close(); close(done) }() + + // Both target writers must enter before either may finish its first Write. awaitStage(t, entered, "first random target Write") awaitStage(t, entered, "second random target Write") release() diff --git a/publication_serial_test.go b/publication_serial_test.go index 779e33d..cfe8805 100644 --- a/publication_serial_test.go +++ b/publication_serial_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "os" + "slices" "sync" "syscall" "testing" @@ -116,15 +117,15 @@ func TestSerialRandomTargetPreservesConflictingPublicationOrder(t *testing.T) { } func TestPublicationReleasesFailedAliasesAfterSourceCache(t *testing.T) { - // Failed outputs before and after a rewrite output retain their shared-path dependency chain. + // Case and decomposition aliases retain their failure chain around a rewrite output. synctest.Test(t, func(t *testing.T) { c := newTestStream(t, WithHashPolicy(HashOff), SetToDevice(DeviceThreads(1))) job := preparedFixture(c, 0, 4096, "same-target", &generatedReader{remaining: 4096}) job.targets = []string{"first", "rewrite", "last"} job.outputs = []targetSpec{ - {name: "first", path: "same-target", device: "fixture"}, - {name: "rewrite", path: "same-target", device: "fixture", output: &fileio.Output{}}, - {name: "last", path: "same-target", device: "fixture"}, + {name: "first", path: "CAF\u00c9", device: "fixture"}, + {name: "rewrite", path: "cafe\u0301", device: "fixture", output: &fileio.Output{}}, + {name: "last", path: "Caf\u00e9", device: "fixture"}, } committed := false c.filesystem = pipelineFilesystem{create: func(_ *writeJob, spec targetSpec) (transferOutput, error) { @@ -147,7 +148,7 @@ func TestPublicationReleasesFailedAliasesAfterSourceCache(t *testing.T) { targets := result.result().Targets if len(targets) != 3 || targets[1].Err != nil || !errors.Is(targets[0].Err, ErrTargetIO) || !errors.Is(targets[2].Err, ErrTargetIO) { - t.Fatalf("alias outcomes: %+v", targets) + t.Errorf("alias outcomes: %+v", targets) } } if count != 1 || !committed { @@ -156,6 +157,135 @@ func TestPublicationReleasesFailedAliasesAfterSourceCache(t *testing.T) { }) } +func TestDirectAliasesWithinItemWaitOnlyForPrecedingItem(t *testing.T) { + for _, test := range []struct { + name string + firstDirect bool + failFirst bool + }{ + {name: "ordinary-first"}, + {name: "direct-first", firstDirect: true}, + {name: "ordinary-create-failure", failFirst: true}, + {name: "direct-create-failure", firstDirect: true, failFirst: true}, + } { + t.Run(test.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // Keep an earlier device lifetime outstanding while the next item prepares aliases. + c := newTestStream(t, WithHashPolicy(HashOff), Overwrite(true), + SetFromDevice(LinearDevice(true)), SetToDevice(DeviceThreads(3))) + gate, holding := make(chan struct{}), make(chan struct{}) + release := onceRelease(gate) + defer release() + created, committed := make(chan string, 3), make(chan string, 3) + c.filesystem = pipelineFilesystem{create: func(job *writeJob, spec targetSpec) (transferOutput, error) { + // Creation failures must release their ticket without stranding the direct sibling. + created <- spec.name + if spec.name == "first" && test.failFirst { + return nil, syscall.EIO + } + written := 0 + return &serialPublicationOutput{ + pipelineOutput: &pipelineOutput{write: func(data []byte) (int, error) { + written += len(data) + return len(data), nil + }}, + sync: func() error { + if job.order == 0 { + close(holding) + <-gate + } + return nil + }, + publish: func(bool) error { + if written != 4096 { + t.Errorf("target %q published %d bytes, want 4096", spec.name, written) + } + committed <- spec.name + return nil + }, + }, nil + }} + preceding := preparedFixture(c, 0, 4096, "preceding", &generatedReader{remaining: 4096}) + preceding.outputs[0].path, preceding.outputs[0].direct = "CAF\u00c9", true + siblings := preparedFixture(c, 1, 4096, "first", &generatedReader{remaining: 4096}) + siblings.targets = []string{"first", "second"} + siblings.outputs = []targetSpec{ + {name: "first", path: "cafe\u0301", device: "fixture", direct: test.firstDirect}, + {name: "second", path: "Caf\u00e9", device: "fixture", direct: true}, + } + prepared := make(chan *writeJob, 2) + prepared <- preceding + prepared <- siblings + close(prepared) + out := c.copy(context.Background(), prepared) + synctest.Wait() + + // Ordinary creation can overlap, but neither direct sibling may open ahead of the predecessor. + select { + case <-holding: + default: + t.Error("preceding output did not reach its held completion") + } + creates := make(map[string]int) + for len(created) > 0 { + creates[<-created]++ + } + wantFirst := 1 + if test.firstDirect { + wantFirst = 0 + } + if creates["preceding"] != 1 || creates["first"] != wantFirst || creates["second"] != 0 || len(committed) != 0 { + t.Errorf("premature direct creation or publication: creates=%v commits=%d", creates, len(committed)) + } + + // Once the prior item settles, both siblings must finish without waiting on their own item. + release() + synctest.Wait() + want := []string{"preceding", "first", "second"} + if test.failFirst { + want = []string{"preceding", "second"} + } + if len(committed) != len(want) { + t.Errorf("siblings stalled after predecessor release: commits=%d, want %d", len(committed), len(want)) + c.stopHard() + } + count := 0 + for job := range out { + count++ + result := job.result() + if result.Err != nil || len(result.Targets) != len(job.targets) { + t.Errorf("unexpected item outcome: %+v", result) + } + for _, target := range result.Targets { + if target.Path == "first" && test.failFirst { + if !errors.Is(target.Err, ErrTargetIO) { + t.Errorf("first creation error=%v, want I/O failure", target.Err) + } + continue + } + if target.Err != nil { + t.Errorf("target %q failed: %v", target.Path, target.Err) + } + } + } + close(created) + for name := range created { + creates[name]++ + } + close(committed) + var publications []string + for name := range committed { + publications = append(publications, name) + } + if count != 2 || creates["preceding"] != 1 || creates["first"] != 1 || creates["second"] != 1 || + !slices.Equal(publications, want) { + t.Errorf("unfinished aliases: results=%d creates=%v publications=%v, want %v", count, creates, publications, want) + } + }) + }) + } +} + func TestLinearSourceKeepsConflictingDirectOutputsExclusive(t *testing.T) { // Raw devices may reject a second Open until the prior Close; Commit cannot hide their writes. synctest.Test(t, func(t *testing.T) { diff --git a/signature_test.go b/signature_test.go index 994b806..91b270d 100644 --- a/signature_test.go +++ b/signature_test.go @@ -1201,7 +1201,8 @@ func (h *blockingHash) Write(p []byte) (int, error) { // holdContentHash makes the run's hash consumer wait on the returned release function before it // hashes its first chunk. It reports when the consumer has started, so a test knows the item's hash -// is being held back, and it restores the pool when the test ends. +// is being held back, and it restores the pool when the test ends. Callers must register later +// cleanup that releases the gates and joins their run before this helper restores the global pool. func holdContentHash(t *testing.T) (started <-chan struct{}, released <-chan struct{}, release func()) { t.Helper() @@ -1220,12 +1221,12 @@ func holdContentHash(t *testing.T) (started <-chan struct{}, released <-chan str } // TestRunHoldsTargetCacheUntilTheHashIsComplete pins the ordering a target's cache entry depends -// on: the writer keeps its descriptor open until the item's hash is complete, so the entry always -// describes the whole content. The run is held back at the hash, and an item that completed anyway +// on: publication waits until the item's hash is complete, so the entry always describes the +// whole content. The run is held back at the hash, and an item that completed anyway // would have published an entry it could not describe. func TestRunHoldsTargetCacheUntilTheHashIsComplete(t *testing.T) { + // Hold a real file's digest while its linear target advances independently. requireSignatureXattrSupport(t) - content := bytes.Repeat([]byte{'c'}, 4096) root := t.TempDir() input := writeSourceFile(t, root, "source.txt", content) @@ -1235,9 +1236,12 @@ func TestRunHoldsTargetCacheUntilTheHashIsComplete(t *testing.T) { // A linear target skips the durability sync, so its writer has nothing left to do but publish // by the time the hash is held back. item := newFixtureItem(input, target) - done := make(chan error, 1) + done := make(chan struct{}) + var runErr error + t.Cleanup(func() { release(); <-done }) go func() { - done <- runFixture( + defer close(done) + runErr = runFixture( context.Background(), newStreamFixture(item), []Item{item}, @@ -1246,6 +1250,7 @@ func TestRunHoldsTargetCacheUntilTheHashIsComplete(t *testing.T) { ) }() + // Confirm the hash gate is occupied before checking target completion. select { case <-started: case <-time.After(10 * time.Second): @@ -1255,8 +1260,8 @@ func TestRunHoldsTargetCacheUntilTheHashIsComplete(t *testing.T) { // The item cannot finish while its hash is unknown, and the writer must be waiting instead of // publishing the entry it cannot describe yet. select { - case err := <-done: - t.Fatalf("the run finished while the item's hash was held back: %v", err) + case <-done: + t.Fatalf("the run finished while the item's hash was held back: %v", runErr) case <-time.After(250 * time.Millisecond): } select { @@ -1268,14 +1273,15 @@ func TestRunHoldsTargetCacheUntilTheHashIsComplete(t *testing.T) { // Releasing the hash lets the item finish and publish the entry of every file it handled. release() select { - case err := <-done: - if err != nil { - t.Fatalf("run error = %v, want nil", err) + case <-done: + if runErr != nil { + t.Fatalf("run error = %v, want nil", runErr) } case <-time.After(10 * time.Second): t.Fatal("the run never finished after the hash was released") } + // Both stored entries must describe the completed content before the result is accepted. if _, err := item.terminal(t); err != nil { t.Fatalf("item failed: %v", err) } diff --git a/target_identity_test.go b/target_identity_test.go index 1d8024a..692b88f 100644 --- a/target_identity_test.go +++ b/target_identity_test.go @@ -161,3 +161,173 @@ func TestTargetParentTraversalFromSymlinkedWorkingDirectory(t *testing.T) { }) } } + +func TestNativeTargetAliasesKeepSerialWinner(t *testing.T) { + // Cover every device configuration that promises ordered completion. + for _, config := range []struct { + name string + sourceLinear, targetLinear bool + targetThreads int + }{ + {"linear-target", false, true, 1}, + {"single-writer", false, false, 1}, + {"linear-source", true, false, 3}, + } { + for _, alias := range []struct { + name string + names [2]string + }{ + {"case", [2]string{"target", "TARGET"}}, + {"unicode-decomposition", [2]string{"caf\u00e9", "cafe\u0301"}}, + } { + t.Run(config.name+"/"+alias.name, func(t *testing.T) { + // Native name lookup determines whether this fixture has one winner or two distinct outputs. + root := t.TempDir() + probe := writeSourceFile(t, root, alias.names[0], []byte("probe")) + first, err := os.Stat(probe) + if err != nil { + t.Fatal(err) + } + second, err := os.Stat(filepath.Join(root, alias.names[1])) + if err != nil && !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + nativeAlias := err == nil && os.SameFile(first, second) + t.Logf("native alias probe: %q and %q alias=%t", alias.names[0], alias.names[1], nativeAlias) + + // Existing exclusive targets are the indexing control; the other cases exercise publication. + for _, test := range []struct { + name string + exists bool + overwrite bool + }{ + {name: "missing-exclusive"}, + {name: "missing-overwrite", overwrite: true}, + {name: "existing-exclusive", exists: true}, + {name: "existing-overwrite", exists: true, overwrite: true}, + } { + t.Run(test.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // Index stable inputs and both native target aliases before any copy changes the destination. + c := newTestStream(t, Overwrite(test.overwrite), WithHashPolicy(HashOff), + SetFromDevice(LinearDevice(config.sourceLinear)), + SetToDevice(LinearDevice(config.targetLinear), DeviceThreads(config.targetThreads))) + dir := filepath.Join(root, test.name) + names := []string{filepath.Join(dir, alias.names[0]), filepath.Join(dir, alias.names[1]), + filepath.Join(dir, "independent")} + if test.exists { + writeSourceFile(t, dir, alias.names[0], []byte("original")) + } + prepared := make(chan *writeJob, len(names)) + for i, name := range names { + content := []string{"first", "second", "independent"}[i] + source := writeSourceFile(t, filepath.Join(dir, "sources"), content, []byte(content)) + job := c.buildJob(&SimpleJob{Path: source, Dsts: []string{name}}, uint64(i)) + if job.itemError != nil { + t.Fatalf("index outcome: %+v", job.result()) + } + wantOutputs := 1 + if i < 2 && test.exists && !test.overwrite && (i == 0 || nativeAlias) { + wantOutputs = 0 + } + if len(job.outputs) != wantOutputs { + t.Fatalf("index order=%d: %+v", i, job.result()) + } + wj := c.prepareItem(job) + t.Cleanup(func() { _ = wj.finishSource() }) + prepared <- wj + } + close(prepared) + + // Hold only the first Commit; native writes, metadata and renames remain real. + gate := make(chan struct{}) + release := onceRelease(gate) + defer release() + native := c.fs() + c.filesystem = pipelineFilesystem{ + transferFilesystem: native, + create: func(job *writeJob, spec targetSpec) (transferOutput, error) { + // Retain native output ownership and delay only the first item's replacement. + out, err := native.Create(job, spec) + if err != nil || job.order != 0 { + return out, err + } + return &delayedPublicationOutput{transferOutput: out, gate: gate}, nil + }, + } + out := c.copy(context.Background(), prepared) + synctest.Wait() + + // A fix must hold the alias while allowing the unrelated file in the same directory to publish. + if data, err := os.ReadFile(names[2]); err != nil || string(data) != "independent" { + t.Errorf("independent output blocked by alias completion: %q / %v", data, err) + } + data, err := os.ReadFile(names[0]) + t.Logf("before first Commit: target=%q error=%v", data, err) + if test.exists { + if err != nil || string(data) != "original" { + t.Errorf("alias published before first Commit: %q / %v", data, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + t.Errorf("missing alias published before first Commit: %q / %v", data, err) + } + release() + + // Drain all real outcomes and retain the first exclusive or last overwrite winner. + count := 0 + for job := range out { + count++ + result := job.result() + if result.Err != nil || len(result.Targets) != 1 || result.Targets[0].Path != names[job.order] { + t.Errorf("unexpected result: %+v", result) + continue + } + targetErr := result.Targets[0].Err + t.Logf("order=%d target error=%v", job.order, targetErr) + refused := test.exists && job.order == 0 || nativeAlias && job.order == 1 + if !test.overwrite && refused { + if !errors.Is(targetErr, os.ErrExist) { + t.Errorf("order=%d: error=%v, want existence failure", job.order, targetErr) + } + } else if targetErr != nil { + t.Errorf("order=%d failed: %v", job.order, targetErr) + } + } + want := "first" + if test.overwrite && nativeAlias { + want = "second" + } else if test.exists && !test.overwrite { + want = "original" + } + data, err = os.ReadFile(names[0]) + t.Logf("final target=%q, want %q; results=%d", data, want, count) + if err != nil || string(data) != want || count != len(names) { + t.Errorf("wrong serial winner: %q / %v, want %q; results=%d", data, err, want, count) + } + if !nativeAlias { + second, err := os.ReadFile(names[1]) + if err != nil || string(second) != "second" { + t.Errorf("distinct native target lost its content: %q / %v", second, err) + } + } + }) + }) + } + }) + } + } +} + +func BenchmarkPublicationOrder(b *testing.B) { + // Measure the per-file completion dependency with ordinary and Unicode target names. + for _, name := range []string{"/volume/archive/file-000001.bin", "/volume/Archive/File-000001.bin", "/volume/caf\u00e9/file-000001.bin"} { + b.Run(name, func(b *testing.B) { + order := newPublicationOrder() + b.ReportAllocs() + for b.Loop() { + turn := order.reserve(name) + turn.release() + } + }) + } +} diff --git a/transfer_target.go b/transfer_target.go index 99e5e89..e225677 100644 --- a/transfer_target.go +++ b/transfer_target.go @@ -5,10 +5,14 @@ import ( "errors" "fmt" "os" + "strings" "sync" "sync/atomic" + "unicode/utf8" mapset "github.com/deckarep/golang-set/v2" + "golang.org/x/text/cases" + "golang.org/x/text/unicode/norm" ) type publicationOrder struct { @@ -17,6 +21,7 @@ type publicationOrder struct { } type publicationTurn struct { + key string previous <-chan struct{} directTurn <-chan struct{} done chan struct{} @@ -27,13 +32,26 @@ func newPublicationOrder() *publicationOrder { return &publicationOrder{pending: make(map[string]chan struct{})} } +func publicationKey(path string) string { + // ASCII paths need only case folding; avoid Unicode transform buffers for ordinary names. + for i := range len(path) { + if path[i] >= utf8.RuneSelf { + return norm.NFD.String(cases.Fold().String(norm.NFD.String(path))) + } + } + return strings.ToLower(path) +} + func (p *publicationOrder) reserve(path string) publicationTurn { - // Only conflicting paths share a completion dependency; finished entries do not accumulate. + // Canonical caseless matching conservatively orders native aliases without changing their paths. + key := publicationKey(path) + + // Only potentially conflicting paths share a dependency; finished entries do not accumulate. p.lock.Lock() defer p.lock.Unlock() done := make(chan struct{}) - turn := publicationTurn{previous: p.pending[path], done: done} - p.pending[path] = done + turn := publicationTurn{key: key, previous: p.pending[key], done: done} + p.pending[key] = done turn.release = func() { // A failed middle output must retain its predecessor's publication dependency. if turn.previous != nil { @@ -44,8 +62,8 @@ func (p *publicationOrder) reserve(path string) publicationTurn { p.lock.Lock() defer p.lock.Unlock() close(done) - if p.pending[path] == done { - delete(p.pending, path) + if p.pending[key] == done { + delete(p.pending, key) } } return turn @@ -57,6 +75,8 @@ func (t *transfer) reservePublications(order *publicationOrder) { if t.noContent || (!c.toDevice.linear && c.toDevice.threads != 1 && !c.fromDevice.linear) { return } + + // Derive each key once and reuse its first predecessor for direct aliases within this item. t.publications = make([]publicationTurn, len(t.job.outputs)) for i, spec := range t.job.outputs { turn := order.reserve(spec.path) @@ -64,8 +84,8 @@ func (t *transfer) reservePublications(order *publicationOrder) { // Aliases in one file share only the prior file's device lifetime, never their own data gate. turn.directTurn = turn.previous for j := 0; j < i; j++ { - if t.job.outputs[j].path == spec.path { - turn.directTurn = t.publications[j].directTurn + if t.publications[j].key == turn.key { + turn.directTurn = t.publications[j].previous break } }