feat: preserve request order for linear targets

Concurrent source preparation can no longer reorder writes to a linear
target. Preparation reports one outcome per indexed request; the ordering
stage buffers later outcomes until the request's turn and only then
transfers the reader, so a source waits for its consumer after the
ordering stage has taken it. Skipped and failed requests advance the order
without a write, so one unresolvable source no longer stalls the requests
behind it. Random targets keep completion-order concurrency.

README.md

@@ -2,6 +2,7 @@
 An Advanced Copy Tools, with following extra features:
 - Process bar
 - Sorted copy order, to improve tape device read performance
+- Concurrent source preparation with request-ordered writes for linear targets
 - Multi target path, read once write many
 - Read file with mmap, with small file prefetch hint
 - JSON format job report

TESTING.md

@@ -79,6 +79,14 @@ Run path and mountpoint tests:
 go test -run '^(TestComparePath|TestSourceRoot|TestFindMountpoint)$' .
 ```
 
+Run linear stream-order tests:
+
+```sh
+go test -run '^(TestRunStreamCopiesRequestsToLinearTarget|TestForwardPreparedOrdersOnlyLinearTargets|TestRunStreamAppliesBoundedBackpressure|TestRunStreamCancellationDrainsPrefetchedJobs|TestLinearTargetStopsWhenDiskUsageEstimateIsInsufficient)$' .
+```
+
+On Linux, include `TestRunStreamMapsDeviceFullToTargetNoSpace` to exercise an actual `/dev/full` write failure.
+
 Run `acp-rewrite` tests:
 
 ```sh

index.go

@@ -34,7 +34,8 @@ func (c *Copyer) index(ctx context.Context) (<-chan *baseJob, error) {
 	go wrap(ctx, func() {
 		defer close(ch)
 
-		for _, job := range jobs {
+		for order, job := range jobs {
+			job.order = uint64(order)
 			select {
 			case <-ctx.Done():
 				return
@@ -52,6 +53,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob {
 		defer close(ch)
 
 		var bytes, files int64
+		var order uint64
 		defer func() {
 			c.submit(&EventUpdateCount{Bytes: bytes, Files: files, Finished: true})
 		}()
@@ -94,6 +96,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob {
 				stat:     stat,
 				targets:  append([]string(nil), request.Targets...),
 				streamID: request.ID,
+				order:    order,
 			}
 			c.submit(&EventUpdateJob{job.report()})
 			bytes += stat.size
@@ -104,6 +107,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob {
 				c.setError(ctx.Err())
 				return
 			case ch <- job:
+				order++
 			}
 		}
 	})

job.go

@@ -42,6 +42,7 @@ type baseJob struct {
 	path     string
 	stat     *stat
 	streamID int64
+	order    uint64
 
 	lock      sync.Mutex
 	writeTime time.Time

prepare.go

@@ -11,6 +11,12 @@ import (
 	"github.com/samuelncui/acp/mmap"
 )
 
+type prepareResult struct {
+	order   uint64
+	job     *writeJob
+	release chan struct{}
+}
+
 func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *writeJob {
 	// Disable read-ahead when a linear source must wait for each consumer.
 	chanLen := 32
@@ -18,21 +24,16 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 		chanLen = 0
 	}
 
-	// Close the prepared stream only after every source worker exits.
-	var wg sync.WaitGroup
-	ch := make(chan *writeJob, chanLen)
-	defer func() {
-		go wrap(ctx, func() {
-			defer close(ch)
-			wg.Wait()
-		})
-	}()
+	// Collect every preparation outcome so skipped jobs can still advance linear ordering.
+	completed := make(chan prepareResult, chanLen)
+	prepared := make(chan *writeJob, chanLen)
+	var workers sync.WaitGroup
 
 	// Prepare source readers with the configured source-device concurrency.
 	for idx := 0; idx < c.fromDevice.threads; idx++ {
-		wg.Add(1)
+		workers.Add(1)
 		go wrap(ctx, func() {
-			defer wg.Done()
+			defer workers.Done()
 
 			// Consume indexed jobs until cancellation or source exhaustion.
 			for {
@@ -43,7 +44,14 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 					if !ok {
 						return
 					}
+					result := prepareResult{order: job.order}
+					if c.toDevice.linear {
+						result.release = make(chan struct{})
+					}
 					if c.linearTargetStopped() {
+						if !sendPrepareResult(ctx, completed, result) {
+							return
+						}
 						continue
 					}
 
@@ -96,16 +104,17 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 							c.reportError(job.path, "", err)
 							job.fail("", err)
 							job.setStatus(jobStatusFinished)
+							if !sendPrepareResult(ctx, completed, result) {
+								return
+							}
 							continue
 						}
 					}
 
-					// Publish the prepared reader and preserve linear-source consumption order.
+					// Transfer the prepared reader to the ordering stage before waiting on a linear source.
 					wj := newWriteJob(job, file, size, c.fromDevice.linear)
-					select {
-					case ch <- wj:
-					case <-ctx.Done():
-						wj.finishSource()
+					result.job = wj
+					if !sendPrepareResult(ctx, completed, result) {
 						return
 					}
 					if !wj.waitConsumed(ctx) {
@@ -116,5 +125,119 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w
 		})
 	}
 
-	return ch
+	// Close preparation outcomes only after every source worker relinquishes ownership.
+	go wrap(ctx, func() {
+		workers.Wait()
+		close(completed)
+	})
+
+	// Preserve request order only where one linear writer requires it.
+	go wrap(ctx, func() {
+		defer close(prepared)
+		c.forwardPrepared(ctx, completed, prepared)
+	})
+	return prepared
+}
+
+func sendPrepareResult(ctx context.Context, completed chan<- prepareResult, result prepareResult) bool {
+	select {
+	case completed <- result:
+	case <-ctx.Done():
+		result.finish()
+		return false
+	}
+	if result.release == nil {
+		return true
+	}
+	select {
+	case <-result.release:
+		return true
+	case <-ctx.Done():
+		return false
+	}
+}
+
+func (result prepareResult) releaseWorker() {
+	if result.release != nil {
+		close(result.release)
+	}
+}
+
+func (result prepareResult) finish() {
+	if result.job != nil {
+		result.job.finishSource()
+	}
+	result.releaseWorker()
+}
+
+func (c *Copyer) forwardPrepared(
+	ctx context.Context,
+	completed <-chan prepareResult,
+	prepared chan<- *writeJob,
+) {
+	// Close readers that remain owned by this stage when cancellation stops forwarding.
+	defer func() {
+		for result := range completed {
+			result.finish()
+		}
+	}()
+
+	// Random targets retain completion-order concurrency.
+	if !c.toDevice.linear {
+		for result := range completed {
+			if result.job == nil {
+				continue
+			}
+			select {
+			case prepared <- result.job:
+			case <-ctx.Done():
+				result.finish()
+				return
+			}
+		}
+		return
+	}
+
+	// A bounded reorder buffer feeds the linear writer in original request order.
+	next := uint64(0)
+	pending := make(map[uint64]prepareResult, c.fromDevice.threads)
+	defer func() {
+		for _, result := range pending {
+			result.finish()
+		}
+	}()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case result, ok := <-completed:
+			if !ok {
+				if len(pending) != 0 {
+					c.setError(fmt.Errorf("ordered preparation ended before request %d", next))
+				}
+				return
+			}
+			pending[result.order] = result
+		}
+
+		for {
+			result, ok := pending[next]
+			if !ok {
+				break
+			}
+			delete(pending, next)
+			next++
+			if result.job == nil {
+				result.releaseWorker()
+				continue
+			}
+			select {
+			case prepared <- result.job:
+				result.releaseWorker()
+			case <-ctx.Done():
+				result.finish()
+				return
+			}
+		}
+	}
 }

stream_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -78,15 +79,10 @@ func TestRunStreamCopiesRequestsToLinearTarget(t *testing.T) {
 	if len(sink.results) != len(source.requests) {
 		t.Fatalf("received %d results, want %d", len(sink.results), len(source.requests))
 	}
-	seen := make(map[int64]struct{}, len(sink.results))
-	for _, result := range sink.results {
-		if result.ID < 1 || result.ID > int64(len(source.requests)) {
-			t.Fatalf("unexpected result ID: %d", result.ID)
+	for index, result := range sink.results {
+		if result.ID != int64(index+1) {
+			t.Fatalf("result %d has ID %d, want %d", index, result.ID, index+1)
 		}
-		if _, exists := seen[result.ID]; exists {
-			t.Fatalf("duplicate result ID: %d", result.ID)
-		}
-		seen[result.ID] = struct{}{}
 		if result.Job.Status != JobStatusFinished || len(result.Job.SuccessTargets) != 1 || result.Job.SHA256 == "" {
 			t.Fatalf("unexpected result: %#v", result.Job)
 		}
@@ -96,6 +92,57 @@ func TestRunStreamCopiesRequestsToLinearTarget(t *testing.T) {
 	}
 }
 
+func TestForwardPreparedOrdersOnlyLinearTargets(t *testing.T) {
+	newJob := func(order uint64, id int64) *writeJob {
+		return newWriteJob(
+			&baseJob{order: order, streamID: id},
+			io.NopCloser(strings.NewReader("fixture")),
+			int64(len("fixture")),
+			false,
+		)
+	}
+	tests := []struct {
+		name    string
+		linear  bool
+		wantIDs []int64
+	}{
+		{name: "linear request order", linear: true, wantIDs: []int64{1, 3}},
+		{name: "random completion order", wantIDs: []int64{3, 1}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Deliver later preparation results first; a failed or skipped middle request has no write Job.
+			copyer := &Copyer{option: &option{
+				fromDevice: &deviceOption{threads: 3},
+				toDevice:   &deviceOption{linear: test.linear},
+			}}
+			completed := make(chan prepareResult, 3)
+			completed <- prepareResult{order: 2, job: newJob(2, 3)}
+			completed <- prepareResult{order: 0, job: newJob(0, 1)}
+			completed <- prepareResult{order: 1}
+			close(completed)
+			prepared := make(chan *writeJob, 3)
+
+			// Forwarding must reorder only the serialized target path.
+			copyer.forwardPrepared(context.Background(), completed, prepared)
+			close(prepared)
+			var ids []int64
+			for job := range prepared {
+				ids = append(ids, job.streamID)
+				job.finishSource()
+			}
+			if len(ids) != len(test.wantIDs) {
+				t.Fatalf("prepared IDs = %v, want %v", ids, test.wantIDs)
+			}
+			for index := range ids {
+				if ids[index] != test.wantIDs[index] {
+					t.Fatalf("prepared IDs = %v, want %v", ids, test.wantIDs)
+				}
+			}
+		})
+	}
+}
+
 func TestRunStreamHashesWithoutTargets(t *testing.T) {
 	// Submit one targetless request through the same bounded stream interface.
 	content := []byte("hash-only fixture")