Method Add

statesync/chunks.go:63–101

Add adds a chunk to the queue. It ignores chunks that already exist, returning false.

(chunk *chunk)

// Add adds a chunk to the queue. It ignores chunks that already exist, returning false.
func (q *chunkQueue) Add(chunk *chunk) (bool, error) {
	if chunk == nil || chunk.Chunk == nil {
		return false, errors.New("cannot add nil chunk")
	}
	q.Lock()
	defer q.Unlock()
	if q.snapshot == nil {
		return false, nil // queue is closed
	}
	if chunk.Height != q.snapshot.Height {
		return false, fmt.Errorf("invalid chunk height %v, expected %v", chunk.Height, q.snapshot.Height)
	}
	if chunk.Format != q.snapshot.Format {
		return false, fmt.Errorf("invalid chunk format %v, expected %v", chunk.Format, q.snapshot.Format)
	}
	if chunk.Index >= q.snapshot.Chunks {
		return false, fmt.Errorf("received unexpected chunk %v", chunk.Index)
	}
	if q.chunkFiles[chunk.Index] != "" {
		return false, nil
	}

	path := filepath.Join(q.dir, strconv.FormatUint(uint64(chunk.Index), 10))
	err := os.WriteFile(path, chunk.Chunk, 0o600)
	if err != nil {
		return false, fmt.Errorf("failed to save chunk %v to file %v: %w", chunk.Index, path, err)
	}
	q.chunkFiles[chunk.Index] = path
	q.chunkSenders[chunk.Index] = chunk.Sender

	// Signal any waiters that the chunk has arrived.
	for _, waiter := range q.waiters[chunk.Index] {
		waiter <- chunk.Index
		close(waiter)
	}
	delete(q.waiters, chunk.Index)

	return true, nil
}

// Allocate allocates a chunk to the caller, making it responsible for fetching it. Returns
// errDone once no chunks are left or the queue is closed.