Repository: prometheus/prometheus
File: tsdb/db.go (Go)

Code before:
	for {
		// If we have a lot of blocks to compact the whole process might take
		// long enough that we end up with a HEAD block that needs to be written.
		// Check if that's the case and stop compactions early.
		if db.head.compactable() && !db.waitingForCompactionDelay() {
			level.Warn(db.logger).Log("msg", "aborting block compactions to persit the head block")
			return nil
		}

		plan, err := db.compactor.Plan(db.dir)
		if err != nil {
			return fmt.Errorf("plan compaction: %w", err)
		}
		if len(plan) == 0 {
			break
		}

		select {
		case <-db.stopc:
			return nil
		default:
		}

		db.mtx.RLock()
		uids, err := db.compactor.Compact(db.dir, plan, db.blocks)
		db.mtx.RUnlock()
		if err != nil {
			return fmt.Errorf("compact %s: %w", plan, err)
		}

		if err := db.reloadBlocks(); err != nil {
			errs := tsdb_errors.NewMulti(fmt.Errorf("reloadBlocks blocks: %w", err))
			for _, uid := range uids {
				if errRemoveAll := os.RemoveAll(filepath.Join(db.dir, uid.String())); errRemoveAll != nil {
					errs.Add(fmt.Errorf("delete persisted block after failed db reloadBlocks:%s: %w", uid, errRemoveAll))
				}
			}
			return errs.Err()
		}
	}

	return nil
}

// getBlock iterates a given block range to find a block by a given id.
// If found it returns the block itself and a boolean to indicate that it was found.
func getBlock(allBlocks []*Block, id ulid.ULID) (*Block, bool) {
	for _, b := range allBlocks {
		if b.Meta().ULID == id {
			return b, true
		}

Review comment (bug):
Compact calls `compactOOOHead` which calls `db.mtx.Lock()`; won't that deadlock?

Code after:

// compactHead compacts the given RangeHead.
// The db.cmtx should be held before calling this method.
func (db *DB) compactHead(head *RangeHead) error {
	uids, err := db.compactor.Write(db.dir, head, head.MinTime(), head.BlockMaxTime(), nil)
	if err != nil {
		return fmt.Errorf("persist head block: %w", err)
	}

	if err := db.reloadBlocks(); err != nil {
		multiErr := tsdb_errors.NewMulti(fmt.Errorf("reloadBlocks blocks: %w", err))
		for _, uid := range uids {
			if errRemoveAll := os.RemoveAll(filepath.Join(db.dir, uid.String())); errRemoveAll != nil {
				multiErr.Add(fmt.Errorf("delete persisted head block after failed db reloadBlocks:%s: %w", uid, errRemoveAll))
			}
		}
		return multiErr.Err()
	}
	if err = db.head.truncateMemory(head.BlockMaxTime()); err != nil {
		return fmt.Errorf("head memory truncate: %w", err)
	}

	db.head.RebuildSymbolTable(db.logger)

	return nil
}

// compactBlocks compacts all the eligible on-disk blocks.
// The db.cmtx should be held before calling this method.
func (db *DB) compactBlocks() (err error) {
	// Check for compactions of multiple blocks.
	for {
		// If we have a lot of blocks to compact the whole process might take
		// long enough that we end up with a HEAD block that needs to be written.
		// Check if that's the case and stop compactions early.
		if db.head.compactable() && !db.waitingForCompactionDelay() {
			db.logger.Warn("aborting block compactions to persit the head block")
			return nil
		}

		plan, err := db.compactor.Plan(db.dir)
		if err != nil {
			return fmt.Errorf("plan compaction: %w", err)
		}
		if len(plan) == 0 {
			break
		}

		select {
		case <-db.stopc:
			return nil