From 4665ef20f952f3c22bcdcc22cbbe1846ecec6385 Mon Sep 17 00:00:00 2001 From: Samuel Cui Date: Fri, 2 Oct 2026 17:14:11 +0800 Subject: [PATCH] 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 | 18 ++++- TESTING.md | 10 +++ acp.go | 24 ++++-- cmd/acp/main.go | 20 +++-- e2e_test.go | 100 +++++++++++++++++++++++++ index.go | 19 +++-- item.go | 8 ++ job.go | 11 +-- prepare.go | 2 +- recover.go | 6 +- stream_test.go | 189 ++++++++++++++++++++++++++++++++++++++++++++++++ 11 files changed, 378 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index a81684d..026e655 100644 --- a/README.md +++ b/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 diff --git a/TESTING.md b/TESTING.md index 057ce8b..50ecd41 100644 --- a/TESTING.md +++ b/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 diff --git a/acp.go b/acp.go index 7f699cf..98f8b4f 100644 --- a/acp.go +++ b/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 diff --git a/cmd/acp/main.go b/cmd/acp/main.go index 2e59d6e..45da10d 100644 --- a/cmd/acp/main.go +++ b/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) } } diff --git a/e2e_test.go b/e2e_test.go index a3162b8..b7ca71c 100644 --- a/e2e_test.go +++ b/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. diff --git a/index.go b/index.go index 6455b62..dacf448 100644 --- a/index.go +++ b/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 { diff --git a/item.go b/item.go index 552a7a2..41458ec 100644 --- a/item.go +++ b/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. diff --git a/job.go b/job.go index 5920f32..0371fe4 100644 --- a/job.go +++ b/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 diff --git a/prepare.go b/prepare.go index d07988b..9d195bd 100644 --- a/prepare.go +++ b/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 } diff --git a/recover.go b/recover.go index 217aa56..82670a3 100644 --- a/recover.go +++ b/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() diff --git a/stream_test.go b/stream_test.go index 0e2c47a..3d98e6f 100644 --- a/stream_test.go +++ b/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) {