mesh-delivery/modules/mesh-delivery-builder/internal/build/places.go

package build

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
)

// DefaultCapacity is the build capacity of a node that has no setting: one build step at a time.
const DefaultCapacity = 1

// ParseCapacity reads the builder's capacity setting: `node=steps` pairs separated by commas, such as
// `shanks=3,ace=2`. A node without a pair has DefaultCapacity. A pair with no node, a count that is not a whole number of
// at least 1, and a node named twice are refused.
func ParseCapacity(s string) (map[string]int, error) {
	out := map[string]int{}
	for _, pair := range strings.Split(s, ",") {
		pair = strings.TrimSpace(pair)
		if pair == "" {
			continue
		}
		node, count, ok := strings.Cut(pair, "=")
		node, count = strings.TrimSpace(node), strings.TrimSpace(count)
		n, err := strconv.Atoi(count)
		if !ok || node == "" || err != nil || n < 1 {
			return nil, fmt.Errorf("the capacity %q is not <node>=<steps at once, at least 1>", pair)
		}
		if _, dup := out[node]; dup {
			return nil, fmt.Errorf("the capacity names %s twice", node)
		}
		out[node] = n
	}
	return out, nil
}

// Places counts the build steps the builder runs on each node and holds a lane back while its node is at its
// capacity (the operator's decision of 2026-10-21; ADR 0478 clause 3). A lane is the fetch and the builds of one build
// module on one node, run one step at a time, so a lane holds one place while it runs. A node is chosen for a lane by
// reserving a place on it at the moment of the choice (choose), so that builds choosing at once see each other's
// choices and none waits on a full node while another has room. A lane waits for its place by watching the ends of the
// other lanes, not by a timer: ending a lane wakes the waiting ones.
type Places struct {
	// Capacity is the number of steps at once by node; a node not in it has DefaultCapacity.
	Capacity map[string]int

	mu       sync.Mutex
	running  map[string]int // lanes holding a place
	reserved map[string]int // lanes chosen for the node that do not hold a place yet
	changed  chan struct{}  // closed, and replaced, each time a lane ends
}

// CapacityOf is how many steps the builder runs on the node at once.
func (p *Places) CapacityOf(node string) int {
	if n, ok := p.Capacity[node]; ok && n >= 1 {
		return n
	}
	return DefaultCapacity
}

// Running is the steps in flight now, by node.
func (p *Places) Running() map[string]int {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make(map[string]int, len(p.running))
	for n, c := range p.running {
		if c > 0 {
			out[n] = c
		}
	}
	return out
}

// Reserved is the lanes chosen for a node that hold no place yet, by node.
func (p *Places) Reserved() map[string]int {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make(map[string]int, len(p.reserved))
	for n, c := range p.reserved {
		if c > 0 {
			out[n] = c
		}
	}
	return out
}

func (p *Places) init() {
	if p.running == nil {
		p.running = map[string]int{}
	}
	if p.reserved == nil {
		p.reserved = map[string]int{}
	}
	if p.changed == nil {
		p.changed = make(chan struct{})
	}
}

// Acquire takes a place on the node, waiting while the node is at its capacity, and answers the function that gives it
// back. A place reserved for the node by a choice is the one taken. It returns an error when ctx ends first, and then
// gives up the reservation too.
func (p *Places) Acquire(ctx context.Context, node string) (release func(), err error) {
	for {
		p.mu.Lock()
		p.init()
		if p.running[node] < p.CapacityOf(node) {
			p.running[node]++
			if p.reserved[node] > 0 {
				p.reserved[node]--
			}
			p.mu.Unlock()
			var once sync.Once
			return func() {
				once.Do(func() {
					p.mu.Lock()
					p.running[node]--
					close(p.changed)
					p.changed = make(chan struct{})
					p.mu.Unlock()
				})
			}, nil
		}
		wait := p.changed
		p.mu.Unlock()
		select {
		case <-wait:
		case <-ctx.Done():
			p.Cancel(node)
			return nil, fmt.Errorf("waiting for a place on %s: %w", node, ctx.Err())
		}
	}
}

// Cancel gives up a place reserved for the node that will not be taken.
func (p *Places) Cancel(node string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.init()
	if p.reserved[node] > 0 {
		p.reserved[node]--
	}
}

// choose orders the nodes (those with room first, counting the lanes running and the lanes reserved, then the fewest of
// them, then by name), keeps the first want and reserves a place on each, all at once.
func (p *Places) choose(nodes []string, want int) []string {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.init()
	load := func(n string) int { return p.running[n] + p.reserved[n] }
	sort.SliceStable(nodes, func(i, j int) bool {
		ri, rj := load(nodes[i]) < p.CapacityOf(nodes[i]), load(nodes[j]) < p.CapacityOf(nodes[j])
		if ri != rj {
			return ri
		}
		if load(nodes[i]) != load(nodes[j]) {
			return load(nodes[i]) < load(nodes[j])
		}
		return nodes[i] < nodes[j]
	})
	if len(nodes) > want {
		nodes = nodes[:want]
	}
	for _, n := range nodes {
		p.reserved[n]++
	}
	return nodes
}