Prepare ACP 0.2.2

Support per-item source read modes and preserve source compatibility.
Fix submission shutdown and accepted-item totals.
Honor CLI index-only operation and report-write failure exit status.

README.md

@@ -61,6 +61,11 @@ type Item interface {
 	                   // hash policy produces a hash, so the default HashOff records no hash
 }
 
+// Optional: override the source device mode for this item.
+type ReadModeItem interface {
+	ReadMode() ReadMode
+}
+
 type SimpleJob struct {
 	Path string
 	Dsts []string
@@ -126,7 +131,9 @@ type TargetResult struct {
    exactly once and the run continues. A panic that escapes pipeline code itself is recorded as
    the run error and ends the run.
 7. **`Close` then `Wait`**, with a single submitter. `Submit` blocks while the read buffer is
-   full, which is the feed's backpressure, and a nil item is a submission error. A caller must
+   full, which is the feed's backpressure, and a nil item is a submission error. When `Close`
+   races a submission, indexed totals include only accepted items, and shutdown waits for the
+   feed's count-event send before closing event delivery. A caller must
    not call `Close`, `Wait` or `Submit` from inside `onResults`: the callback runs on the
    delivery goroutine, so `Close` would wait for the call it is in.
 
@@ -195,7 +202,10 @@ concurrently, so it may keep unguarded state; registering the same handler twice
 registration, so it is still called once. The results callback is called from one goroutine as
 well, never concurrently.
 
-`WithReadMode` applies to the source device only. Buffered reads are the default and avoid
+`WithReadMode` applies to the source device only. An item implementing `ReadModeItem` overrides
+that choice for its own source; ACP calls the method once during indexing, and an invalid mode
+or panic fails that item without stopping the run. Items without the method keep the device mode.
+Buffered reads are the default and avoid
 updating the source access time where the platform allows it (`O_NOATIME`, with a fallback when
 the open is refused); mapped reads use the `mmap` package, whose reader reports the end of an
 empty mapping immediately. Either way one descriptor serves the whole item.
@@ -446,7 +456,9 @@ Usage of acp:
 
 The command exits `0` only when every selected item was copied to every requested target.
 Any item failure, any target that was not written, and any pipeline failure make it exit
-non-zero, and the report file is written either way.
+non-zero, and the report file is written either way. Failure to write a requested report also
+exits non-zero. `-notarget` selects an index operation with no destinations, including when
+`-target` flags were supplied.
 
 ## Example
 

TESTING.md

@@ -101,6 +101,9 @@ non-zero. `TestACPCommandExitsNonZeroWhenACopyFails` refuses one target with `-n
 a non-zero exit status together with a report that names the refused target.
 
 The command end-to-end tests are skipped when `go test` is run with `-short`.
+`TestACPCommandHonorsIndexAndReportFlags` also verifies that `-notarget` leaves an explicit
+target untouched and reports no target outcomes, and that a completed copy exits non-zero
+when its requested report cannot be written.
 
 ## Focused tests
 
@@ -184,6 +187,13 @@ mid-run:
 go test -run '^(TestRunStopsFeedingItemsAfterGracefulStop|TestRunDoesNotReportAStopAfterACompleteRun|TestRunCancellationDrainsPrefetchedItems|TestRunReportsEveryAcceptedItemExactlyOnce)$' .
 ```
 
+Run feed shutdown tests, which hold the count-event handoff while `Close` waits, verify the
+concurrent feed counters, and exclude an item whose handoff was refused from indexed totals:
+
+```sh
+go test -race -run '^(TestCloseWaitsForTheSubmissionCountEvent|TestSubmitCountsOnlyAcceptedItems|TestSubmitRacingCloseKeepsTheFeedCountersSafe)$' .
+```
+
 Run linear stream-order tests:
 
 ```sh

acp.go

@@ -162,6 +162,7 @@ func newStream(ctx context.Context, onResults func([]Result) error, opt *option)
 // submission error that does not end the feed: it is recorded as the run's error, so Wait reports
 // it, and the items submitted after it are still accepted.
 func (c *StreamCopyer) Submit(items ...Item) error {
+	// Check stopping reasons once per batch, so accepted work can still drain after a stop.
 	if err := c.feedStop(); err != nil {
 		// An exhausted linear target stays an item outcome: the items that reached it carry
 		// the error, and the run itself is not a failure.
@@ -171,6 +172,7 @@ func (c *StreamCopyer) Submit(items ...Item) error {
 		return err
 	}
 
+	// Describe and hand off each item before reporting the batch's indexed totals.
 	for _, item := range items {
 		if item == nil {
 			err := fmt.Errorf("submit failed, item is nil")
@@ -181,17 +183,19 @@ func (c *StreamCopyer) Submit(items ...Item) error {
 		// Orders are zero-based: the linear reorder buffer forwards from request zero.
 		job := c.buildJob(item, c.order)
 		c.order++
-		if job.itemError == nil && job.stat != nil {
-			c.countFiles.Add(1)
-			c.countBytes.Add(job.stat.size)
-		}
 		if err := c.push(job); err != nil {
 			c.setError(err)
 			return err
 		}
 	}
-	c.submit(&EventUpdateCount{Bytes: c.countBytes.Load(), Files: c.countFiles.Load()})
 
+	// Close must not end the pipeline while the feed still owns an event send. If it already
+	// closed the feed, the pipeline's final count event owns the totals instead.
+	c.feedLock.Lock()
+	defer c.feedLock.Unlock()
+	if !c.closed {
+		c.submit(&EventUpdateCount{Bytes: c.countBytes.Load(), Files: c.countFiles.Load()})
+	}
 	return nil
 }
 
@@ -227,15 +231,23 @@ func (c *StreamCopyer) Wait() error {
 // push hands one job to the read buffer. It holds the feed lock, so Close cannot close the
 // channel between the check and the send.
 func (c *StreamCopyer) push(job *baseJob) error {
+	// Keep channel ownership with the feed until this handoff has settled.
 	c.feedLock.Lock()
 	defer c.feedLock.Unlock()
-
 	if c.closed {
 		return errStreamClosed
 	}
 
+	// Snapshot indexed facts before another stage can update the job, and count them only if
+	// the handoff succeeds, before Close can publish the pipeline's final totals.
+	var files, size int64
+	if job.itemError == nil && job.stat != nil {
+		files, size = 1, job.stat.size
+	}
 	select {
 	case c.readCh <- job:
+		c.countFiles.Add(files)
+		c.countBytes.Add(size)
 		return nil
 	case <-c.hardStop:
 		return errStreamStopped

cmd/acp/main.go

@@ -34,18 +34,21 @@ func init() {
 }
 
 func main() {
+	// Parse the command and establish its cancellation context.
 	ctx, cancel := context.WithCancel(context.Background())
-
 	cpuid.Flags()
 	flag.Parse()
 	cpuid.Detect()
 
+	// An index has no destinations even when -target was also supplied.
 	sources := flag.Args()
 	if len(sources) == 0 {
 		logrus.Fatalf("cannot found source path")
 	}
 
-	if !*noTarget && len(targetPaths) == 0 {
+	if *noTarget {
+		targetPaths = nil
+	} else if len(targetPaths) == 0 {
 		targetPaths = append(targetPaths, sources[len(sources)-1])
 		sources = sources[:len(sources)-1]
 	}
@@ -53,6 +56,7 @@ func main() {
 		logrus.Fatalf("cannot found source path")
 	}
 
+	// Interrupts stop the feed while the pipeline drains accepted work.
 	signals := make(chan os.Signal, 1)
 	signal.Notify(signals, os.Interrupt)
 	go func() {
@@ -64,6 +68,7 @@ func main() {
 		}
 	}()
 
+	// Collect the report and configure the requested copy or index operation.
 	report := newReport()
 	opts := make([]acp.Option, 0, 8)
 
@@ -79,6 +84,7 @@ func main() {
 	opts = append(opts, acp.WithHash(*reportPath != ""))
 	opts = append(opts, acp.Overwrite(!*notOverwrite))
 
+	// Linear devices serialize their own side of the transfer.
 	if *fromLinear {
 		opts = append(opts, acp.SetFromDevice(acp.LinearDevice(true)))
 	}
@@ -101,6 +107,7 @@ func main() {
 	}
 	opts = append(opts, acp.WithEventHandler(handler))
 
+	// Drain the run before deciding its status or writing the report.
 	copyer, err := acp.New(ctx, opts...)
 	if err != nil {
 		logrus.Fatalf("unexpected exit: %s", err)
@@ -112,14 +119,15 @@ func main() {
 
 	// The report is stored even when the run failed, so a batch can be inspected after the
 	// command told its caller that something went wrong.
-	if err := storeReport(report, *reportPath, *reportIndent); err != nil {
-		logrus.Warnf("open report fail, path= '%s', err= %s", *reportPath, err)
+	reportErr := storeReport(report, *reportPath, *reportIndent)
+	if reportErr != nil {
+		logrus.Warnf("open report fail, path= '%s', err= %s", *reportPath, reportErr)
 		logrus.Infof("report= %q", report.getter().ToJSONString(false))
 	}
 
 	// A copy that did not finish is not a success: an item ACP could not process, a target
-	// that was not written, or a pipeline failure all make the command exit non-zero.
-	if runErr != nil || report.hasFailure() {
+	// that was not written, a pipeline failure or an unsaved requested report all exit non-zero.
+	if runErr != nil || reportErr != nil || report.hasFailure() {
 		os.Exit(1)
 	}
 }

e2e_test.go

@@ -162,6 +162,106 @@ func TestACPCommandE2E(t *testing.T) {
 	}
 }
 
+func TestACPCommandHonorsIndexAndReportFlags(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping end-to-end test in short mode")
+	}
+
+	// Build the command once so both cases exercise its actual flag parsing and exit status.
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	repoRoot, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	binary := filepath.Join(t.TempDir(), "acp")
+	if runtime.GOOS == "windows" {
+		binary += ".exe"
+	}
+	build := exec.CommandContext(ctx, "go", "build", "-o", binary, "./cmd/acp")
+	build.Dir = repoRoot
+	if output, err := build.CombinedOutput(); err != nil {
+		t.Fatalf("build the command: %v\n%s", err, output)
+	}
+
+	t.Run("notarget overrides explicit targets", func(t *testing.T) {
+		// An explicit target already contains data that an index operation must leave alone.
+		root := t.TempDir()
+		source := filepath.Join(root, "source.txt")
+		target := filepath.Join(root, "target")
+		if err := os.MkdirAll(target, 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(source, []byte("source content"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		existing := filepath.Join(target, "source.txt")
+		if err := os.WriteFile(existing, []byte("existing content"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+
+		// Request the index together with -target: it must still have no target outcomes.
+		reportPath := filepath.Join(root, "index.json")
+		command := exec.CommandContext(ctx, binary, "-p=false", "-notarget", "-target", target, "-report", reportPath, source)
+		command.Dir = repoRoot
+		if output, err := command.CombinedOutput(); err != nil {
+			t.Fatalf("run the index: %v\n%s", err, output)
+		}
+		data, err := os.ReadFile(reportPath)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var report acp.Report
+		if err := json.Unmarshal(data, &report); err != nil {
+			t.Fatal(err)
+		}
+		if len(report.Errors) != 0 || len(report.Jobs) != 1 {
+			t.Fatalf("index errors/rows = %d/%d, want 0/1", len(report.Errors), len(report.Jobs))
+		}
+		row := report.Jobs[0]
+		if row.SHA256 == "" || len(row.SuccessTargets) != 0 || len(row.FailTargets) != 0 {
+			t.Errorf("index row = %#v, want a hash and no targets", row)
+		}
+		content, err := os.ReadFile(existing)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(content) != "existing content" {
+			t.Errorf("existing target = %q, want unchanged content", content)
+		}
+	})
+
+	t.Run("report failure exits nonzero", func(t *testing.T) {
+		// A directory cannot be replaced by a report file, regardless of host permissions.
+		root := t.TempDir()
+		source := filepath.Join(root, "source.txt")
+		target := filepath.Join(root, "target.txt")
+		reportPath := filepath.Join(root, "report-directory")
+		if err := os.WriteFile(source, []byte("fixture"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.Mkdir(reportPath, 0o755); err != nil {
+			t.Fatal(err)
+		}
+
+		// A successful copy with an unsaved requested report must tell its caller it failed.
+		command := exec.CommandContext(ctx, binary, "-p=false", "-report", reportPath, source, target)
+		command.Dir = repoRoot
+		output, err := command.CombinedOutput()
+		var exitErr *exec.ExitError
+		if !errors.As(err, &exitErr) || !strings.Contains(string(output), "open report fail") {
+			t.Errorf("command exit = %v, want a report-write failure:\n%s", err, output)
+		}
+		content, err := os.ReadFile(target)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(content) != "fixture" {
+			t.Errorf("target = %q, want a completed copy", content)
+		}
+	})
+}
+
 // selectedSourceFiles returns every regular file below source, which is the set of files a
 // wildcard run selects. The pre-stream API exposed that selection as data; the shell does not,
 // so the test walks the tree itself instead of asking the library what it selected.

index.go

@@ -60,11 +60,12 @@ func (c *StreamCopyer) newJob(item Item, targets []string, order uint64) (*baseJ
 
 	path := filepath.Clean(sourceName)
 	job := &baseJob{
-		copyer:  c,
-		item:    item,
-		path:    path,
-		targets: targets,
-		order:   order,
+		copyer:   c,
+		item:     item,
+		path:     path,
+		targets:  targets,
+		order:    order,
+		readMode: c.fromDevice.readMode,
 	}
 
 	// A transfer always reads its source, so a policy that trades a computed hash for a
@@ -75,6 +76,14 @@ func (c *StreamCopyer) newJob(item Item, targets []string, order uint64) (*baseJ
 			c.hashPolicy, path,
 		)
 	}
+	if selected, ok := item.(ReadModeItem); ok {
+		if err := protectCall("Item.ReadMode", func() { job.readMode = selected.ReadMode() }); err != nil {
+			return job, err
+		}
+		if job.readMode != ReadBuffered && job.readMode != ReadMapped {
+			return job, fmt.Errorf("unknown item read mode, mode= %s", job.readMode)
+		}
+	}
 
 	info, err := os.Stat(path)
 	if err != nil {

item.go

@@ -23,6 +23,14 @@ type Item interface {
 	Targets() []string
 }
 
+// ReadModeItem optionally selects how one item's source content is read. Items that do not
+// implement it use the source device mode set by WithReadMode, which defaults to ReadBuffered.
+// ACP calls ReadMode once while indexing a describable item. A panic or invalid mode fails only
+// that item; it does not stop the run.
+type ReadModeItem interface {
+	ReadMode() ReadMode
+}
+
 // SimpleJob is an Item that carries no state of its own: Path is the source and Dsts are the
 // destinations. A caller that submits a SimpleJob finds its result by matching Result.Job
 // against the pointer it submitted, because SimpleJob has no callback of its own.

job.go

@@ -24,11 +24,12 @@ const (
 )
 
 type baseJob struct {
-	copyer *StreamCopyer
-	item   Item
-	path   string
-	stat   *stat
-	order  uint64
+	copyer   *StreamCopyer
+	item     Item
+	path     string
+	stat     *stat
+	order    uint64
+	readMode ReadMode
 
 	// itemError reports an item ACP could not process at all, instead of a completion.
 	itemError error

prepare.go

@@ -72,7 +72,7 @@ func (c *StreamCopyer) prepareItem(job *baseJob) *writeJob {
 	}
 
 	// Only an item that reads content opens a mapping: a reuse-only item reads a stored hash.
-	mode := c.fromDevice.readMode
+	mode := job.readMode
 	if !needsContent {
 		mode = ReadBuffered
 	}

recover.go

@@ -38,9 +38,9 @@ func panicError(what string, value any) error {
 }
 
 // protectCall runs one call into caller-implemented code: the results callback, an event handler,
-// or Item.Source/Item.Targets. A panic becomes an error instead of unwinding a pipeline goroutine,
-// because an unwound goroutine keeps publishing on channels it no longer owns and leaves the caller
-// without a terminal outcome.
+// or Item.Source/Item.Targets/ReadModeItem.ReadMode. A panic becomes an error instead of
+// unwinding a pipeline goroutine, because an unwound goroutine keeps publishing on channels it
+// no longer owns and leaves the caller without a terminal outcome.
 func protectCall(what string, f func()) (err error) {
 	defer func() {
 		e := recover()

stream_test.go

@@ -36,6 +36,136 @@ func collectResults() (func([]Result) error, func() [][]Result) {
 	return callback, snapshot
 }
 
+type readModeItem struct {
+	*SimpleJob
+	mode     ReadMode
+	panicErr error
+	calls    int
+}
+
+func (i *readModeItem) ReadMode() ReadMode {
+	i.calls++
+	if i.panicErr != nil {
+		panic(i.panicErr)
+	}
+	return i.mode
+}
+
+func TestItemsOverrideTheStreamReadMode(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		streamMode ReadMode
+		withOption bool
+		overrides  []ReadMode
+	}{
+		{name: "implicit buffered default", overrides: []ReadMode{ReadMapped, ReadBuffered}},
+		{name: "mapped stream default", streamMode: ReadMapped, withOption: true, overrides: []ReadMode{ReadBuffered, ReadMapped}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			root := t.TempDir()
+			plain := &SimpleJob{Path: writeSourceFile(t, root, "plain", []byte("plain")), Dsts: []string{filepath.Join(root, "plain-copy")}}
+			first := &readModeItem{SimpleJob: &SimpleJob{Path: writeSourceFile(t, root, "first", []byte("first")), Dsts: []string{filepath.Join(root, "first-copy")}}, mode: tc.overrides[0]}
+			second := &readModeItem{SimpleJob: &SimpleJob{Path: writeSourceFile(t, root, "second", []byte("second")), Dsts: []string{filepath.Join(root, "second-copy")}}, mode: tc.overrides[1]}
+			items := []Item{plain, first, second}
+
+			// Observe the mode ACP actually passes to its source opener, including the item that
+			// has no override. Source preparation may run concurrently.
+			previous := openSourceContent
+			var lock sync.Mutex
+			opened := make(map[string]ReadMode)
+			openSourceContent = func(path string, mode ReadMode) (itemSource, error) {
+				lock.Lock()
+				opened[path] = mode
+				lock.Unlock()
+				return previous(path, mode)
+			}
+			t.Cleanup(func() { openSourceContent = previous })
+
+			callback, batches := collectResults()
+			opts := []Option{WithHashPolicy(HashReadRefresh)}
+			if tc.withOption {
+				opts = append(opts, SetFromDevice(WithReadMode(tc.streamMode)))
+			}
+			if err := runStream(context.Background(), callback, items, opts...); err != nil {
+				t.Fatalf("run error = %v", err)
+			}
+
+			wantModes := map[string]ReadMode{plain.Path: tc.streamMode, first.Path: first.mode, second.Path: second.mode}
+			lock.Lock()
+			if !reflect.DeepEqual(opened, wantModes) {
+				t.Errorf("source modes = %v, want %v", opened, wantModes)
+			}
+			lock.Unlock()
+			if first.calls != 1 || second.calls != 1 {
+				t.Errorf("override calls = %d/%d, want one each", first.calls, second.calls)
+			}
+			results := make(map[Item]Result)
+			for _, batch := range batches() {
+				for _, result := range batch {
+					results[result.Job] = result
+				}
+			}
+			if len(results) != len(items) {
+				t.Fatalf("results = %d, want %d", len(results), len(items))
+			}
+			for _, item := range items {
+				result := results[item]
+				if result.Err != nil || len(result.Targets) != 1 || result.Targets[0].Err != nil {
+					t.Fatalf("item %q result = %#v", item.Source(), result)
+				}
+				got, err := os.ReadFile(item.Targets()[0])
+				if err != nil {
+					t.Fatal(err)
+				}
+				want, err := os.ReadFile(item.Source())
+				if err != nil {
+					t.Fatal(err)
+				}
+				if !bytes.Equal(got, want) {
+					t.Errorf("target %q = %q, want %q", item.Targets()[0], got, want)
+				}
+			}
+		})
+	}
+}
+
+func TestItemReadModeFailureDoesNotStopTheRun(t *testing.T) {
+	root := t.TempDir()
+	panicCause := errors.New("read mode fixture")
+	panicking := &readModeItem{SimpleJob: &SimpleJob{Path: writeSourceFile(t, root, "panic", []byte("panic"))}, panicErr: panicCause}
+	invalid := &readModeItem{SimpleJob: &SimpleJob{Path: writeSourceFile(t, root, "invalid", []byte("invalid"))}, mode: ReadMode(99)}
+	good := &SimpleJob{Path: writeSourceFile(t, root, "good", []byte("good"))}
+	callback, batches := collectResults()
+	if err := runStream(context.Background(), callback, []Item{panicking, invalid, good}, WithHashPolicy(HashRead)); err != nil {
+		t.Fatalf("run error = %v, want item failures only", err)
+	}
+
+	results := make(map[Item]Result)
+	for _, batch := range batches() {
+		for _, result := range batch {
+			if _, exists := results[result.Job]; exists {
+				t.Fatalf("item %q was reported twice", result.Job.Source())
+			}
+			results[result.Job] = result
+		}
+	}
+	if len(results) != 3 {
+		t.Fatalf("results = %d, want 3", len(results))
+	}
+	if err := results[panicking].Err; !errors.Is(err, panicCause) || !strings.Contains(err.Error(), "Item.ReadMode panicked") {
+		t.Errorf("panicking item error = %v, want protected caller panic", err)
+	}
+	if err := results[invalid].Err; err == nil || !strings.Contains(err.Error(), "unknown item read mode") {
+		t.Errorf("invalid item error = %v, want invalid mode", err)
+	}
+	if result := results[good]; result.Err != nil || len(result.SHA256) != sha256.Size {
+		t.Errorf("following item result = %#v, want a complete hash", result)
+	}
+	if panicking.calls != 1 || invalid.calls != 1 {
+		t.Errorf("override calls = %d/%d, want one each", panicking.calls, invalid.calls)
+	}
+}
+
 // TestFailedResultBypassesTheResultBuffer pins error rule 1: a result that carries an error is
 // delivered immediately instead of waiting for a batch or the flush interval, so a caller can
 // persist it before it submits the next batch. Both error outlets count: the item's own error and
@@ -694,6 +824,65 @@ func TestSubmitRacingCloseKeepsTheFeedCountersSafe(t *testing.T) {
 	}
 }
 
+func TestCloseWaitsForTheSubmissionCountEvent(t *testing.T) {
+	// Hold the event handoff after the item has entered the read buffer, just as a blocked
+	// event handler can. Close must not let the pipeline close that handoff under its sender.
+	input := writeSourceFile(t, t.TempDir(), "source.txt", []byte("fixture"))
+	stream := newTestStream(t, WithReadBuffer(1))
+	stream.eventCh = make(chan Event)
+	submitted := make(chan error, 1)
+	go func() { submitted <- stream.Submit(&SimpleJob{Path: input}) }()
+	<-stream.readCh
+	select {
+	case err := <-submitted:
+		t.Fatalf("Submit() returned %v before its count event was received", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	// Shutdown must wait until the feed relinquishes its count-event send.
+	closed := make(chan error, 1)
+	go func() { closed <- stream.Close() }()
+	returnedEarly := false
+	select {
+	case <-closed:
+		returnedEarly = true
+	case <-time.After(50 * time.Millisecond):
+	}
+	<-stream.eventCh
+	if err := <-submitted; err != nil {
+		t.Fatalf("Submit() error = %v", err)
+	}
+	if !returnedEarly {
+		if err := <-closed; err != nil {
+			t.Fatalf("Close() error = %v", err)
+		}
+	}
+	if returnedEarly {
+		t.Fatal("Close returned while the feed still owned a blocked count-event send")
+	}
+}
+
+func TestSubmitCountsOnlyAcceptedItems(t *testing.T) {
+	// The first item fills the read buffer; a fatal stop then refuses the next handoff.
+	input := writeSourceFile(t, t.TempDir(), "source.txt", []byte("fixture"))
+	stream := newTestStream(t, WithReadBuffer(1))
+	if err := stream.Submit(&SimpleJob{Path: input}); err != nil {
+		t.Fatal(err)
+	}
+	stream.stopHard()
+	if err := stream.Submit(&SimpleJob{Path: input}); !errors.Is(err, errStreamStopped) {
+		t.Fatalf("Submit() error = %v, want the fatal-stop refusal", err)
+	}
+
+	// Indexed totals describe accepted work, including when the feed loses a handoff.
+	if files, size := stream.countFiles.Load(), stream.countBytes.Load(); files != 1 || size != int64(len("fixture")) {
+		t.Fatalf("indexed files/bytes = %d/%d, want 1/%d", files, size, len("fixture"))
+	}
+	if err := stream.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // TestItemTargetsPanicBecomesThatItemsResultError pins the second half of the item-panic outlet:
 // a panic in Item.Targets is that item's error, exactly like a panic in Item.Source.
 func TestItemTargetsPanicBecomesThatItemsResultError(t *testing.T) {