fix: preserve linear target no-space failures

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)

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))
 				}
 			}

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() {

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)
+	}
+}

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)
+	}
+}

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 {

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)
 	}

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)