acp/transfer_read.go

package acp

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
)

func (t *transfer) readContent() {
	// Every exit closes both data channels, releases the source turn and starts one source Close.
	defer func() {
		if value := recover(); value != nil {
			t.readErr = t.panicFailure("source reader", value)
		}
		// An Open error or panic still releases writers waiting for observed metadata.
		if t.sourceReady != nil {
			t.sourceReadyOnce.Do(func() { close(t.sourceReady) })
		}
		if t.job.endpoints && t.readErr != nil {
			t.stopEndpoint(t.readErr)
		}
		close(t.chunks)
		if t.hashes != nil {
			close(t.hashes)
		}
		t.ahead.finishRead(t.job.order)
		close(t.readDone)
		go t.copyer.wrap(context.Background(), func() {
			defer close(t.sourceDone)
			defer func() {
				if value := recover(); value != nil {
					t.sourceErr = t.panicFailure("source close", value)
				}
			}()
			t.sourceErr = t.job.finishSource()
			if t.job.contentSource != nil {
				t.stopEndpoint(t.sourceErr)
			}
		})
	}()
	if !t.wait(t.readTurn) || t.noContent {
		return
	}

	// Observe the acquired file once before allowing any output allocation or header emission.
	if t.job.contentSource != nil {
		var file fs.File
		var err error
		panicErr := protectCall("ContentSource.Open", func() {
			file, err = t.job.contentSource.Open(t.copyer.ctx)
		})
		t.job.reader = file
		err = errors.Join(err, panicErr, t.copyer.ctx.Err())
		if err == nil && file == nil {
			err = fmt.Errorf("content source returned no file")
		}
		if err == nil {
			var metadata ContentMetadata
			panicErr = protectCall("content source metadata", func() {
				metadata.Info, err = file.Stat()
				err = errors.Join(err, t.copyer.ctx.Err())
				if err != nil {
					return
				}
				if attrs, ok := file.(XattrReader); ok {
					metadata.Xattrs, err = attrs.Xattrs()
					err = errors.Join(err, t.copyer.ctx.Err())
					if err != nil {
						return
					}
				}
				t.job.metadata, err = snapshotContentMetadata(metadata)
			})
			err = errors.Join(err, panicErr, t.copyer.ctx.Err())
		}
		if err != nil {
			t.readErr, t.job.itemError = err, err
			return
		}
		t.job.stat = contentStat(t.job.metadata)
		t.sourceReadyOnce.Do(func() { close(t.sourceReady) })
		if t.job.stat.size == 0 {
			return
		}
	}

	// Quota-limited short blocks are not EOF; only the actual reader outcome ends content.
	var offset int64
	for !t.stopped() {
		eof := t.job.stat != nil && offset >= t.job.stat.size
		want, err := t.ahead.wait(t.job.order, t.start+offset, batchSize, eof, t.stop)
		if err != nil {
			t.readErr = err
			return
		}
		if t.stopped() {
			break
		}
		n, done, err := t.readChunk(want)
		offset += int64(n)
		if err != nil {
			t.readErr = err
			return
		}
		if done {
			if t.job.endpoints && offset != t.job.stat.size {
				t.readErr = fmt.Errorf("content size mismatch, read=%d, expected=%d: %w",
					offset, t.job.stat.size, io.ErrUnexpectedEOF)
			}
			return
		}
	}
	t.readErr = t.stoppedError()
}

func (t *transfer) readChunk(want int) (int, bool, error) {
	// Backing ownership is established before reading; no read slot is held during queue waits.
	chunk, err := t.ahead.acquire(t.job.order, t.stop)
	if err != nil {
		return 0, false, err
	}
	defer chunk.release()
	release, err := t.ahead.readSlot(t.job.order, t.stop)
	if err != nil {
		return 0, false, err
	}
	var n int
	func() {
		defer release()
		n, err = io.ReadFull(t.job.reader, chunk.data[:want])
	}()
	done := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)
	if err != nil && !done {
		return 0, false, fmt.Errorf("read source failed, %w", err)
	}
	if n == 0 {
		return 0, done, nil
	}

	// Hash and target consumers own independent references to the same immutable bytes.
	chunk.data = chunk.data[:n]
	consumers := [2]chan *chunkBuffer{t.chunks, t.hashes}
	for _, dst := range consumers {
		if dst == nil {
			continue
		}
		reference := chunk.retain()
		select {
		case dst <- reference:
		case <-t.stop:
			reference.release()
			return n, false, t.stoppedError()
		case <-t.copyer.hardStop:
			reference.release()
			return n, false, t.stoppedError()
		}
	}
	return n, done, nil
}