Method prefetch

internal/stream/pipe.go:142–242

prefetch runs in a goroutine, fetching blocks concurrently and sending to blockQueue.

()

// prefetch runs in a goroutine, fetching blocks concurrently and sending to blockQueue.
func (p *StreamPipe) prefetch() {
	defer close(p.blockQueue)

	// calc block boundaries
	alignedStart := p.start - (p.start % p.blockSize)
	leftTrim := p.start - alignedStart
	rightTrim := (p.end % p.blockSize) + 1
	totalBlocks := int((p.end - alignedStart + p.blockSize) / p.blockSize)

	currentBlock := 0
	offset := alignedStart

	for currentBlock < totalBlocks {
		// check for cancellation
		select {
		case <-p.ctx.Done():
			return
		default:
		}

		// fetch a batch of blocks concurrently
		batchSize := min(config.ValueOf.StreamConcurrency, totalBlocks-currentBlock)
		blocks := make([][]byte, batchSize)

		var wg sync.WaitGroup
		var fetchErr error
		var errMu sync.Mutex

		for i := range batchSize {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()

				blockNum := currentBlock + idx
				blockOffset := offset + int64(idx)*p.blockSize

				data, err := p.downloadBlockWithRetry(blockOffset)
				dataLen := int64(len(data))

				if err != nil {
					errMu.Lock()
					if fetchErr == nil {
						fetchErr = err
					}
					errMu.Unlock()
					return
				}

				// trim first/last block to exact range
				if totalBlocks == 1 {
					if dataLen < rightTrim {
						rightTrim = dataLen
					}
					if leftTrim > dataLen {
						leftTrim = dataLen
					}
					data = data[leftTrim:rightTrim]
				} else if blockNum == 0 {