diff --git a/acp.go b/acp.go index bc97ba1..9842a86 100644 --- a/acp.go +++ b/acp.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "github.com/sirupsen/logrus" ) @@ -16,6 +17,7 @@ type Copyer struct { eventCh chan Event getDevice func(in string) string getDiskUsageCache func(mountPoint string) *diskUsageCache + linearTargetEnded uint32 } func New(ctx context.Context, opts ...Option) (*Copyer, error) { @@ -74,6 +76,17 @@ func (c *Copyer) setError(err error) { } } +func (c *Copyer) endLinearTarget(err error) { + if !c.toDevice.linear || !checkErrorAbort(err) { + return + } + atomic.StoreUint32(&c.linearTargetEnded, 1) +} + +func (c *Copyer) linearTargetStopped() bool { + return atomic.LoadUint32(&c.linearTargetEnded) != 0 +} + func (c *Copyer) run(ctx context.Context) error { defer c.running.Done() defer close(c.eventCh) diff --git a/cleanup.go b/cleanup.go index e078c24..47cedfe 100644 --- a/cleanup.go +++ b/cleanup.go @@ -14,8 +14,10 @@ func (c *Copyer) cleanupJob(ctx context.Context, copyed <-chan *baseJob) bool { return streamSinkFailed } - for _, dst := range job.successTargets { - if err := writeSysStat(dst, job.stat); err != nil { + 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)) } } diff --git a/copy.go b/copy.go index 02da853..80f70e2 100644 --- a/copy.go +++ b/copy.go @@ -68,6 +68,10 @@ func (c *Copyer) copy(ctx context.Context, prepared <-chan *writeJob) <-chan *ba job.finishSource() continue } + if c.linearTargetStopped() { + job.finishSource() + continue + } wrap(ctx, func() { c.write(ctx, job, ch, cntr, noSpaceDevices) }) } @@ -120,19 +124,22 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c continue } - if err := c.getDiskUsageCache(dev).check(job.size); err != nil { - if errors.Is(err, ErrTargetNoSpace) { - noSpaceDevices.Add(dev) - } + if !c.toDevice.linear { + if err := c.getDiskUsageCache(dev).check(job.size); err != nil { + if errors.Is(err, ErrTargetNoSpace) { + noSpaceDevices.Add(dev) + } - job.fail(target, fmt.Errorf("check disk usage have error, %w", err)) - continue + job.fail(target, fmt.Errorf("check disk usage have error, %w", err)) + continue + } } if err := mappingError(os.MkdirAll(filepath.Dir(target), os.ModePerm)); err != nil { if checkErrorAbort(err) { noSpaceDevices.Add(dev) } + c.endLinearTarget(err) job.fail(target, fmt.Errorf("mkdir dst dir fail, %w", err)) continue @@ -143,6 +150,7 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c if checkErrorAbort(err) { noSpaceDevices.Add(dev) } + c.endLinearTarget(err) job.fail(target, fmt.Errorf("open dst file fail, %w", err)) continue @@ -170,20 +178,20 @@ func (c *Copyer) write(ctx context.Context, job *writeJob, ch chan<- *baseJob, c return } - // avoid block channel - for range ch { - } - - if err := os.Remove(target); err != nil { - c.reportError(job.path, target, fmt.Errorf("delete failed file has error, %w", err)) - } - rerr = mappingError(rerr) if checkErrorAbort(rerr) { noSpaceDevices.Add(dev) } + c.endLinearTarget(rerr) + + // avoid block channel + for range ch { + } job.fail(target, fmt.Errorf("write dst file fail, %w", rerr)) + if err := os.Remove(target); err != nil { + c.reportError(job.path, target, fmt.Errorf("delete failed file has error, %w", err)) + } }() defer func() { diff --git a/copy_test.go b/copy_test.go index c63e5ac..4b0e8bd 100644 --- a/copy_test.go +++ b/copy_test.go @@ -3,9 +3,11 @@ package acp import ( "bytes" "context" + "errors" "io" "os" "path/filepath" + "syscall" "testing" "time" @@ -170,3 +172,76 @@ func TestWriteReturnsWhenCanceledBeforePublishing(t *testing.T) { t.Fatalf("reader closed %d times, want 1", reader.closed) } } + +func TestBaseJobFailureMovesSuccessfulTarget(t *testing.T) { + // Seed a completed target before applying a metadata-stage failure. + copyer := &Copyer{option: newOption(), eventCh: make(chan Event, 2)} + job := &baseJob{ + copyer: copyer, src: &source{}, stat: &stat{}, successTargets: []string{"target"}, + } + + // A late target failure must remove the target from the successful result. + job.fail("target", syscall.ENOSPC) + report := job.report() + if len(report.SuccessTargets) != 0 { + t.Fatalf("success targets = %v, want none", report.SuccessTargets) + } + if !errors.Is(report.FailTargets["target"], syscall.ENOSPC) { + t.Fatalf("target failure = %v, want %v", report.FailTargets["target"], syscall.ENOSPC) + } +} + +func TestFirstTargetFailureRemainsAuthoritative(t *testing.T) { + // Record a mapped write failure before the best-effort cleanup error. + copyer := &Copyer{option: newOption(), eventCh: make(chan Event, 4)} + job := &baseJob{copyer: copyer, src: &source{}, stat: &stat{}} + job.fail("target", mappingError(syscall.ENOSPC)) + copyer.setError(errors.New("remove failed")) + + // Secondary cleanup failures must not hide the no-space classification. + if err := copyer.WaitErr(); !errors.Is(err, ErrTargetNoSpace) { + t.Fatalf("WaitErr() = %v, want %v", err, ErrTargetNoSpace) + } +} + +func TestLinearTargetSkipsDiskUsageEstimate(t *testing.T) { + // Build one linear write whose disk-usage lookup would fail the test if called. + root := t.TempDir() + target := filepath.Join(root, "target") + copyer := &Copyer{ + option: newOption(), eventCh: make(chan Event, 8), + getDevice: func(string) string { return root }, + getDiskUsageCache: func(string) *diskUsageCache { + t.Fatal("linear target queried filesystem capacity") + return nil + }, + } + copyer.toDevice.linear = true + job := newWriteJob(&baseJob{ + copyer: copyer, src: &source{}, path: "source", stat: &stat{size: 1}, targets: []string{target}, + }, io.NopCloser(bytes.NewReader([]byte("x"))), 1, false) + completed := make(chan *baseJob, 1) + + // The write must reach the target without consulting statfs-derived capacity. + copyer.write(context.Background(), job, completed, new(counter), mapset.NewSet[string]()) + report := (<-completed).report() + if len(report.SuccessTargets) != 1 || report.SuccessTargets[0] != target { + t.Fatalf("success targets = %v, want %q", report.SuccessTargets, target) + } +} + +func TestStoppedLinearTargetDoesNotReadStreamSource(t *testing.T) { + // Mark a linear target exhausted before its stream indexer requests more work. + source := new(sliceStreamSource) + copyer := &Copyer{option: newOption(), eventCh: make(chan Event, 2)} + copyer.streamSource = source + copyer.toDevice.linear = true + copyer.endLinearTarget(ErrTargetNoSpace) + + // A stopped target closes the index stream without consuming another request. + for range copyer.indexStream(context.Background()) { + } + if source.index != 0 { + t.Fatalf("source requests = %d, want 0", source.index) + } +} diff --git a/full_linux_test.go b/full_linux_test.go new file mode 100644 index 0000000..f06c714 --- /dev/null +++ b/full_linux_test.go @@ -0,0 +1,36 @@ +//go:build linux + +package acp + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestRunStreamMapsDeviceFullToTargetNoSpace(t *testing.T) { + // Route a disposable target symlink to Linux's deterministic ENOSPC device. + root := t.TempDir() + input := filepath.Join(root, "source") + target := filepath.Join(root, "target") + if err := os.WriteFile(input, []byte("fixture"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.Symlink("/dev/full", target); err != nil { + t.Fatal(err) + } + source := &sliceStreamSource{requests: []*StreamRequest{{ + ID: 1, Source: input, Targets: []string{target}, + }}} + + // The synchronous stream boundary must preserve the portable no-space sentinel. + err := RunStream( + context.Background(), source, new(collectingStreamSink), + Overwrite(true), SetToDevice(LinearDevice(true)), + ) + if !errors.Is(err, ErrTargetNoSpace) { + t.Fatalf("RunStream() error = %v, want %v", err, ErrTargetNoSpace) + } +} diff --git a/index.go b/index.go index 04cecc7..d96a939 100644 --- a/index.go +++ b/index.go @@ -56,6 +56,9 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob { c.submit(&EventUpdateCount{Bytes: bytes, Files: files, Finished: true}) }() for { + if c.linearTargetStopped() { + return + } request, err := c.streamSource.Next(ctx) if err != nil { if err != io.EOF { diff --git a/job.go b/job.go index 2478562..27fdf21 100644 --- a/job.go +++ b/job.go @@ -86,6 +86,13 @@ func (j *baseJob) fail(path string, err error) { j.lock.Lock() defer j.lock.Unlock() + for index, target := range j.successTargets { + if target != path { + continue + } + j.successTargets = append(j.successTargets[:index], j.successTargets[index+1:]...) + break + } if j.failedTargets == nil { j.failedTargets = make(map[string]error, 1) } diff --git a/prepare.go b/prepare.go index 1a3deea..ec4f99e 100644 --- a/prepare.go +++ b/prepare.go @@ -39,6 +39,9 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w if !ok { return } + if c.linearTargetStopped() { + continue + } job.setStatus(jobStatusPreparing)