mesh-controller/cmd/mesh-controller/composer.go

package main

import (
	"context"
	"errors"
	"sort"
	"sync"
)

// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
type composer struct {
	composeNodes func(ctx context.Context, nodes []string) map[string]error
	writeGrants  func(ctx context.Context) error

	mu      sync.Mutex
	pending map[string]bool
	grants  bool
	waiters []waiter
	wake    chan struct{}
}

// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
// goes.
type waiter struct {
	nodes  []string
	grants bool
	done   chan passResult
}

// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them.
type passResult struct {
	failed map[string]error
	grants error
}

func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
	return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
}

// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
	failed, grantsErr, err := c.composeEach(ctx, nodes, grants)
	if err != nil {
		return err
	}
	var errs []error
	for _, n := range nodes {
		if e := failed[n]; e != nil {
			errs = append(errs, e)
		}
	}
	return errors.Join(append(errs, grantsErr)...)
}

// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one
// pass and settles each module by its own nodes); err is ctx's when it ends first.
func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) {
	done := make(chan passResult, 1)
	c.mu.Lock()
	for _, n := range nodes {
		c.pending[n] = true
	}
	c.grants = c.grants || grants
	c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
	c.mu.Unlock()
	select {
	case c.wake <- struct{}{}:
	default:
	}
	select {
	case r := <-done:
		return r.failed, r.grants, nil
	case <-ctx.Done():
		return nil, nil, ctx.Err()
	}
}

// run makes the passes until ctx ends: each takes everything asked for since the last began.
func (c *composer) run(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.wake:
		}
		c.mu.Lock()
		nodes := make([]string, 0, len(c.pending))
		for n := range c.pending {
			nodes = append(nodes, n)
		}
		grants, waiters := c.grants, c.waiters
		c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
		c.mu.Unlock()
		if len(waiters) == 0 {
			continue
		}
		sort.Strings(nodes)
		var failed map[string]error
		if len(nodes) > 0 {
			failed = c.composeNodes(ctx, nodes)
		}
		// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
		// own callers alone (issue 1945).
		var grantsErr error
		if grants {
			grantsErr = c.writeGrants(ctx)
		}
		for _, w := range waiters {
			r := passResult{failed: map[string]error{}}
			for _, n := range w.nodes {
				if err := failed[n]; err != nil {
					r.failed[n] = err
				}
			}
			if w.grants {
				r.grants = grantsErr
			}
			w.done <- r
		}
	}
}