fix: harden stream failure recovery

2026-09-03 22:57:23 +00:00 · 2026-08-28 16:08:34 +08:00
parent bc886ff506
commit 4eecd3fe31
6 changed files with 186 additions and 10 deletions
@@ -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))
@@ -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
 		}
 	}
 }
@@ -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)
+	}
+}
@@ -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) {
@@ -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)
@@ -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():