diff --git a/acp.go b/acp.go index 9842a86..d19b050 100644 --- a/acp.go +++ b/acp.go @@ -88,20 +88,28 @@ func (c *Copyer) linearTargetStopped() bool { } func (c *Copyer) run(ctx context.Context) error { + // Give internal failures one cancellation boundary without canceling the caller. + ctx, cancel := context.WithCancel(ctx) + defer cancel() defer c.running.Done() defer close(c.eventCh) + // Keep event dispatch alive until every pipeline stage stops publishing. go wrap(ctx, func() { c.eventLoop(ctx) }) + // Start the bounded index before connecting downstream stages. indexed, err := c.index(ctx) if err != nil { c.setError(err) return err } + // Run preparation, copying, and result persistence as one pipeline. prepared := c.prepare(ctx, indexed) copyed := c.copy(ctx, prepared) - sinkFailed := c.cleanupJob(ctx, copyed) + sinkFailed := c.cleanupJob(ctx, cancel, copyed) + + // Flush persisted results unless a Sink write already failed. if c.streamSink != nil && !sinkFailed { if err := c.streamSink.Flush(ctx); err != nil { c.setError(fmt.Errorf("flush stream sink failed, %w", err)) diff --git a/cleanup.go b/cleanup.go index 47cedfe..602140b 100644 --- a/cleanup.go +++ b/cleanup.go @@ -2,36 +2,47 @@ package acp import ( "context" + "errors" "fmt" + "os" ) -func (c *Copyer) cleanupJob(ctx context.Context, copyed <-chan *baseJob) bool { - streamSinkFailed := false +func (c *Copyer) cleanupJob(ctx context.Context, cancel context.CancelFunc, copyed <-chan *baseJob) bool { for { select { case job, ok := <-copyed: if !ok { - return streamSinkFailed + return false } + // Restore metadata before publishing the final result. for _, dst := range append([]string(nil), job.successTargets...) { if err := mappingError(writeSysStat(dst, job.stat)); err != nil { c.endLinearTarget(err) job.fail(dst, fmt.Errorf("change info, write sys stat fail, %w", err)) c.reportError(job.path, dst, fmt.Errorf("change info, write sys stat fail, %w", err)) + + // Remove the failed target so the same operation can retry it. + if err := os.Remove(dst); err != nil && !errors.Is(err, os.ErrNotExist) { + c.reportError(job.path, dst, fmt.Errorf("delete target after metadata failure failed, %w", err)) + } } } + // Publish only results whose data and metadata lifecycle has finished. job.setStatus(jobStatusFinished) - if c.streamSink != nil && !streamSinkFailed { - if err := c.streamSink.Write(ctx, &StreamResult{ID: job.streamID, Job: job.report()}); err != nil { - c.setError(fmt.Errorf("write stream result failed, id=%d, %w", job.streamID, err)) - streamSinkFailed = true - } + if c.streamSink == nil { + continue + } + if err := c.streamSink.Write(ctx, &StreamResult{ID: job.streamID, Job: job.report()}); err != nil { + // Stop upstream writes once final results can no longer be persisted. + c.setError(fmt.Errorf("write stream result failed, id=%d, %w", job.streamID, err)) + cancel() + return true } case <-ctx.Done(): c.setError(ctx.Err()) - return streamSinkFailed + return false } } } diff --git a/cleanup_test.go b/cleanup_test.go new file mode 100644 index 0000000..ebccccf --- /dev/null +++ b/cleanup_test.go @@ -0,0 +1,59 @@ +package acp + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestCleanupRemovesTargetAfterMetadataFailure(t *testing.T) { + // Use a dangling symlink so metadata restoration fails while removal remains possible. + root := t.TempDir() + sourcePath := filepath.Join(root, "source") + if err := os.WriteFile(sourcePath, []byte("fixture"), 0o644); err != nil { + t.Fatal(err) + } + info, err := os.Stat(sourcePath) + if err != nil { + t.Fatal(err) + } + stat, err := newStat(sourcePath, info) + if err != nil { + t.Fatal(err) + } + target := filepath.Join(root, "target") + if err := os.Symlink(filepath.Join(root, "missing"), target); err != nil { + t.Skipf("create test symlink: %v", err) + } + + // Cleanup must discard the failed target before publishing its final result. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + option := newOption() + if err := option.check(); err != nil { + t.Fatal(err) + } + copyer := &Copyer{option: option, eventCh: make(chan Event, 8)} + job := &baseJob{ + copyer: copyer, + src: &source{base: root, path: "source"}, + path: sourcePath, + stat: stat, + successTargets: []string{target}, + } + copyed := make(chan *baseJob, 1) + copyed <- job + close(copyed) + if sinkFailed := copyer.cleanupJob(ctx, cancel, copyed); sinkFailed { + t.Fatal("cleanup reported an unexpected Sink failure") + } + if _, err := os.Lstat(target); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("stat failed target error = %v, want %v", err, os.ErrNotExist) + } + report := job.report() + if !errors.Is(report.FailTargets[target], os.ErrNotExist) { + t.Fatalf("metadata failure = %v, want %v", report.FailTargets[target], os.ErrNotExist) + } +} diff --git a/copy.go b/copy.go index b842ac2..c7a09e3 100644 --- a/copy.go +++ b/copy.go @@ -82,6 +82,7 @@ func (c *Copyer) copy(ctx context.Context, prepared <-chan *writeJob) <-chan *ba } func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, cntr *counter, noSpaceDevices mapset.Set[string]) { + // Release the source and publish ownership only after every consumer stops. var wg sync.WaitGroup defer func() { wg.Wait() @@ -93,6 +94,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c } }() + // Reject source changes before creating any targets. job.setStatus(jobStatusCopying) if job.size != job.stat.size { job.fail("", fmt.Errorf("source size changed, indexed=%d current=%d", job.stat.size, job.size)) @@ -106,6 +108,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c return } + // Track progress and close every consumer after the source reader finishes. atomic.AddInt64(&cntr.files, 1) chans := make([]chan []byte, 0, len(job.targets)+1) defer func() { @@ -114,10 +117,12 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c } }() + // Open each viable target before reading the source once. var readErr error for _, target := range job.targets { target := target + // Reject exhausted targets before reserving capacity. dev := c.getDevice(target) if noSpaceDevices.Contains(dev) { job.fail(target, ErrTargetNoSpace) @@ -134,6 +139,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c continue } + // Prepare the target file before attaching its stream consumer. if err := mappingError(os.MkdirAll(filepath.Dir(target), os.ModePerm)); err != nil { if checkErrorAbort(err) { noSpaceDevices.Add(dev) @@ -163,6 +169,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c } } + // Consume source buffers in one managed writer for this target. ch := make(chan []byte, 4) chans = append(chans, ch) @@ -170,6 +177,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c go wrap(ctx, func() { defer wg.Done() + // Settle target status and discard any incomplete file before exiting. var rerr error defer func() { if rerr == nil { @@ -193,6 +201,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c } }() + // Write every source buffer before publishing the durability boundary. defer func() { if file != nil { _ = file.Close() @@ -229,6 +238,9 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c } }) } + targetWriters := len(chans) + + // Add hashing as another consumer of the shared source stream. if c.withHash { sha := sha256Pool.Get().(hash.Hash) sha.Reset() @@ -248,6 +260,8 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c job.setHash(sha.Sum(nil)) }) } + + // Read the source only when at least one target or hash consumer needs it. if len(chans) == 0 { return } @@ -256,6 +270,9 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c if readErr == nil && copied != job.size { readErr = fmt.Errorf("source size changed while copying, expected=%d copied=%d", job.size, copied) } + if readErr != nil && targetWriters == 0 { + job.fail("", readErr) + } } func (c *Copyer) streamCopy(ctx context.Context, dsts []chan []byte, src io.ReadCloser, bytes *int64) (int64, error) { diff --git a/copy_test.go b/copy_test.go index 7c8a950..064b2a1 100644 --- a/copy_test.go +++ b/copy_test.go @@ -29,6 +29,18 @@ func (r *trackingReadCloser) Close() error { return nil } +type failingReadCloser struct { + err error +} + +func (r *failingReadCloser) Read([]byte) (int, error) { + return 0, r.err +} + +func (*failingReadCloser) Close() error { + return nil +} + func TestCopyEmptyFile(t *testing.T) { tests := []struct { name string @@ -112,6 +124,30 @@ func TestWritePublishesFinishingJob(t *testing.T) { } } +func TestHashOnlyReadFailureFailsJob(t *testing.T) { + // Build a target-free hash Job whose source fails on its first read. + readErr := errors.New("read failed") + copyer := &Copyer{option: newOption(), eventCh: make(chan Event, 8)} + copyer.withHash = true + job := newWriteJob(&baseJob{ + copyer: copyer, + src: &source{}, + path: "source", + stat: &stat{size: 1}, + }, &failingReadCloser{err: readErr}, 1, false) + completed := make(chan *baseJob, 1) + + // The source error must cross both the Job result and synchronous error boundary. + copyer.write(context.Background(), job, completed, new(counter), mapset.NewSet[string]()) + report := (<-completed).report() + if !errors.Is(report.FailTargets[""], readErr) { + t.Fatalf("hash-only failure = %v, want %v", report.FailTargets[""], readErr) + } + if err := copyer.WaitErr(); !errors.Is(err, readErr) { + t.Fatalf("WaitErr() = %v, want %v", err, readErr) + } +} + func TestWriteJobWaitConsumedReturnsOnCancellation(t *testing.T) { // Model a linear source whose reader has already moved to the Copy stage. job := newWriteJob(nil, new(trackingReadCloser), 0, true) diff --git a/stream_test.go b/stream_test.go index 2c41d01..a25cdee 100644 --- a/stream_test.go +++ b/stream_test.go @@ -260,6 +260,51 @@ type untilCanceledStreamSource struct { nextID int64 } +type blockingStreamSource struct { + request *StreamRequest + calls int +} + +func (s *blockingStreamSource) Next(ctx context.Context) (*StreamRequest, error) { + s.calls++ + if s.calls == 1 { + return s.request, nil + } + + <-ctx.Done() + return nil, ctx.Err() +} + +func TestRunStreamCancelsSourceAfterSinkFailure(t *testing.T) { + // Let the source block on its next request after producing one copy Job. + root := t.TempDir() + input := filepath.Join(root, "source.txt") + if err := os.WriteFile(input, []byte("fixture"), 0o644); err != nil { + t.Fatal(err) + } + source := &blockingStreamSource{request: &StreamRequest{ + ID: 1, Source: input, Targets: []string{filepath.Join(root, "target.txt")}, + }} + sinkErr := errors.New("sink failed") + + // A Sink error must cancel the blocked Source instead of copying unpersisted work forever. + done := make(chan error, 1) + go func() { + done <- RunStream(context.Background(), source, &collectingStreamSink{err: sinkErr}) + }() + select { + case err := <-done: + if !errors.Is(err, sinkErr) { + t.Fatalf("RunStream() error = %v, want %v", err, sinkErr) + } + case <-time.After(5 * time.Second): + t.Fatal("RunStream did not cancel its Source after the Sink failed") + } + if source.calls != 2 { + t.Fatalf("source calls = %d, want 2", source.calls) + } +} + func (s *untilCanceledStreamSource) Next(ctx context.Context) (*StreamRequest, error) { select { case <-ctx.Done():