tar/asm: simplify done handling

Refactor `newInputTarStreamCommon` to accept an optional
`done chan<- error` instead of a `withDone` bool.

Clarify documentation to state that callers must fully read the returned
reader.

Signed-off-by: Jan Kaluza <jkaluza@redhat.com>

Author: jankaluza

tar/asm/disassemble.go

@@ -161,8 +161,8 @@ func newInputTarStreamCommon(
 	r io.Reader,
 	p storage.Packer,
 	fp storage.FilePutter,
-	withDone bool,
-) (pr *io.PipeReader, done <-chan error) {
+	done chan<- error,
+) (pr *io.PipeReader) {
 	// What to do here... folks will want their own access to the Reader that is
 	// their tar archive stream, but we'll need that same stream to use our
 	// forked 'archive/tar'.
@@ -181,22 +181,14 @@ func newInputTarStreamCommon(
 	// `archive/tar` doesn't care.
 	pr, pw := io.Pipe()
 
-	// We need a putter that will generate the crc64 sums of file payloads.
 	if fp == nil {
 		fp = storage.NewDiscardFilePutter()
 	}
 
 	outputRdr := io.TeeReader(r, pw)
+	go runInputTarStreamGoroutine(outputRdr, pw, p, fp, done)
 
-	if withDone {
-		ch := make(chan error, 1)
-		done = ch
-		go runInputTarStreamGoroutine(outputRdr, pw, p, fp, ch)
-		return pr, done
-	}
-
-	go runInputTarStreamGoroutine(outputRdr, pw, p, fp, nil)
-	return pr, nil
+	return pr
 }
 
 // NewInputTarStream wraps the Reader stream of a tar archive and provides a
@@ -213,9 +205,9 @@ func newInputTarStreamCommon(
 // If callers need to be able to abort early and/or wait for goroutine termination,
 // prefer NewInputTarStreamWithDone.
 //
-// Deprecated: Use NewInputTarStreamWithDone instead.
+// Deprecated: This leaves a goroutine around if the consumer aborts without consuming.
 func NewInputTarStream(r io.Reader, p storage.Packer, fp storage.FilePutter) (io.Reader, error) {
-	pr, _ := newInputTarStreamCommon(r, p, fp, false)
+	pr := newInputTarStreamCommon(r, p, fp, nil)
 	return pr, nil
 }
 
@@ -232,7 +224,12 @@ func NewInputTarStream(r io.Reader, p storage.Packer, fp storage.FilePutter) (io
 // The returned reader is an io.ReadCloser so callers can stop early; closing it
 // aborts the pipe so the internal goroutine can terminate promptly (rather than
 // hanging on a blocked pipe write).
+//
+// The caller is expected to consume the returned reader fully until EOF
+// (not just the tar EOF marker); closing the returned reader earlier will
+// cause the done channel to return a failure.
 func NewInputTarStreamWithDone(r io.Reader, p storage.Packer, fp storage.FilePutter) (io.ReadCloser, <-chan error, error) {
-	pr, done := newInputTarStreamCommon(r, p, fp, true)
+	done := make(chan error, 1)
+	pr := newInputTarStreamCommon(r, p, fp, done)
 	return pr, done, nil
 }

tar/asm/disassemble_test.go

@@ -97,10 +97,10 @@ func (p *recordingPacker) AddEntry(e storage.Entry) (int, error) {
 	// Copy payload because callers may reuse buffers.
 	cp := e
 	if e.Payload != nil {
-		cp.Payload = append([]byte(nil), e.Payload...)
+		cp.Payload = bytes.Clone(e.Payload)
 	}
 	p.entries = append(p.entries, cp)
-	return len(cp.Payload), nil
+	return len(p.entries), nil
 }
 
 func (p *recordingPacker) snapshot() []storage.Entry {
@@ -118,17 +118,18 @@ type recordingFilePutter struct {
 }
 
 func (fp *recordingFilePutter) Put(name string, r io.Reader) (int64, []byte, error) {
-	b, err := io.ReadAll(r)
+	dataLen, err := io.Copy(io.Discard, r)
 	if err != nil {
 		return 0, nil, err
 	}
+
 	fp.mu.Lock()
 	fp.puts = append(fp.puts, name)
 	fp.mu.Unlock()
 
 	// Return a deterministic "checksum" based on content length.
-	csum := []byte(fmt.Sprintf("len=%d", len(b)))
-	return int64(len(b)), csum, nil
+	csum := []byte(fmt.Sprintf("len=%d", dataLen))
+	return dataLen, csum, nil
 }
 
 // Helper function to generate the tar with optional extra padding.
@@ -339,7 +340,8 @@ func TestNewInputTarStreamWithDonUnderlyingClosed(t *testing.T) {
 		t.Fatalf("timeout waiting for underlying reader to enter blocked state")
 	}
 
-	// Now "close" the underlying reader while the tar-split goroutine is still running.
+	// Now trigger an error from the underlying closableBlockingReader while the tar-split
+	// goroutine is still running.
 	under.Close()
 
 	// The tar-split goroutine should treat this as a non-EOF error, call fail(err),