Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/fixed-14645.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
- The external-mutation boundary now gives each issue or pull request its own writer slot instead of one slot per repository, so label writes no longer queue behind a review, merge or comment on a different item; the per-target limit is configurable via `convergence.mutation.max_writers_per_target` (default 1) ([#14645](https://github.com/hivecommons/hive/issues/14645)).
3 changes: 2 additions & 1 deletion src/cmd/hive/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -2024,7 +2024,8 @@ func (b *boot) bootAdvisoryWith(deps bootAdvisoryDeps) bool {
stats := &effects.Recorder{}
stats.SetMode(mode)
b.mutationStats = stats
ledger, err := mutation.OpenLedger(filepath.Join(mutationStateDir, "claims.json"), mutation.DefaultMaxWritersPerRepo)
ledger, err := mutation.OpenLedgerScoped(filepath.Join(mutationStateDir, "claims.json"),
b.cfg.MutationMaxWritersPerTarget(), mutation.TargetWriterScope)
if err != nil {
b.logger.Error("mutation convergence ledger unavailable; external mutation fencing disabled", "mode", mode, "error", err)
return false
Expand Down
22 changes: 22 additions & 0 deletions src/pkg/config/convergence.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,28 @@ type ConvergenceConfig struct {
// future build -- still resolves to "off": an unparseable setting must
// never select a posture the operator did not ask for.
Mode string `yaml:"mode,omitempty" json:"mode,omitempty"`
// Mutation tunes the external-mutation boundary (pkg/convergence/mutation).
Mutation *ConvergenceMutationConfig `yaml:"mutation,omitempty" json:"mutation,omitempty"`
}

// ConvergenceMutationConfig tunes the external-mutation boundary's claim
// ledger.
type ConvergenceMutationConfig struct {
// MaxWritersPerTarget is how many external mutations may be in flight at
// once against one issue or pull request (owner/repo#N); mutations with
// no number (branch pushes, issue/PR creation) share one pool per repo.
// Zero or unset selects the default, 1 (hivecommons/hive#14645).
MaxWritersPerTarget int `yaml:"max_writers_per_target,omitempty" json:"max_writers_per_target,omitempty"`
}

// MutationMaxWritersPerTarget returns the configured
// convergence.mutation.max_writers_per_target, or 0 (the ledger default) when
// unset, non-positive, or c is nil.
func (c *Config) MutationMaxWritersPerTarget() int {
if c == nil || c.Convergence.Mutation == nil || c.Convergence.Mutation.MaxWritersPerTarget <= 0 {
return 0
}
return c.Convergence.Mutation.MaxWritersPerTarget
}

const (
Expand Down
29 changes: 28 additions & 1 deletion src/pkg/config/convergence_test.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,10 @@
package config

import "testing"
import (
"testing"

"gopkg.in/yaml.v3"
)

// ── kubestellar/hive#3845 convergence feature toggle ───────────────────────────
//
Expand Down Expand Up @@ -103,3 +107,26 @@ func TestConvergenceMode_EnvOverridesConfig(t *testing.T) {
t.Fatalf("env=enforce over config=shadow resolved %q, want enforce", got)
}
}

// TestMutationMaxWritersPerTarget pins the #14645 knob: unset/non-positive
// selects the ledger default (0), a positive value is passed through.
func TestMutationMaxWritersPerTarget(t *testing.T) {
var nilCfg *Config
if got := nilCfg.MutationMaxWritersPerTarget(); got != 0 {
t.Fatalf("nil config = %d, want 0", got)
}
if got := (&Config{}).MutationMaxWritersPerTarget(); got != 0 {
t.Fatalf("unset = %d, want 0", got)
}
neg := &Config{Convergence: ConvergenceConfig{Mutation: &ConvergenceMutationConfig{MaxWritersPerTarget: -3}}}
if got := neg.MutationMaxWritersPerTarget(); got != 0 {
t.Fatalf("negative = %d, want 0", got)
}
var cfg Config
if err := yaml.Unmarshal([]byte("convergence:\n mutation:\n max_writers_per_target: 3\n"), &cfg); err != nil {
t.Fatal(err)
}
if got := cfg.MutationMaxWritersPerTarget(); got != 3 {
t.Fatalf("configured = %d, want 3", got)
}
}
8 changes: 5 additions & 3 deletions src/pkg/convergence/mutation/boundary.go
Original file line number Diff line number Diff line change
Expand Up @@ -102,9 +102,11 @@ func (b *Boundary) Execute(ctx context.Context, claim effects.Claim, effect func
})
// The claim guards the window of the external call only. Every exit path
// must release it: a claim left ActiveMutation after a refused or failed
// effect holds the repo's writer slot until TTL, and at enforce that
// denies every other mutation on the repo for ten minutes (observed on a
// live hive as "writers exhausted (max 1)" behind one stuck review).
// effect holds its writer slot until TTL, and at enforce that denies
// every other mutation in the same writer scope for ten minutes (observed
// on a live hive as "repo ... writers exhausted (max 1)" behind one stuck
// review while the slot was per repo; with TargetWriterScope it is
// "target owner/repo#N writers exhausted" and only that item waits).
b.release(executor, claim, mclaim, entry)
logicalID := mutEffect.LogicalID()
if execErr != nil {
Expand Down
99 changes: 88 additions & 11 deletions src/pkg/convergence/mutation/ledger.go
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
package mutation

import (
Expand All @@ -7,6 +7,7 @@
"os"
"path/filepath"
"sort"
"strings"
"sync"
"syscall"
"time"
Expand All @@ -33,7 +34,8 @@
// ErrClaimHeld: an overlapping claim is actively held; the waiter
// consumes no capacity and simply does not acquire.
ErrClaimHeld = errors.New("an overlapping mutation claim is actively held")
// ErrNoCapacity: the repository's mutation-writer slots are all consumed.
// ErrNoCapacity: the writer scope's mutation-writer slots are all
// consumed (the repository by default; see WriterScope).
ErrNoCapacity = errors.New("no mutation-writer capacity is available")
// ErrStaleEpoch: the presented epoch is not the entry's current epoch, or
// the entry is not in a state that epoch may act in. Every stale-owner
Expand All @@ -54,6 +56,52 @@
// mutation-writer slot per canonical repository.
const DefaultMaxWritersPerRepo = 1

// DefaultMaxWritersPerTarget is the default capacity of a TargetWriterScope
// ledger: one mutation-writer slot per target (repo#number).
const DefaultMaxWritersPerTarget = 1

// WriterScope maps a claim to the capacity bucket its mutation-writer slot is
// counted in. Claims whose scopes differ never compete for capacity (they
// still conflict through Overlaps).
type WriterScope func(Claim) string

// RepoWriterScope counts every claim against its canonical repository: one
// shared pool of slots per repo. It is the OpenLedger default.
func RepoWriterScope(c Claim) string { return c.Repo }

// TargetWriterScope counts a claim against the issue or pull request it
// mutates ("owner/repo#N"), so a label write on one item never waits behind a
// review, merge or comment on another (hivecommons/hive#14645). It
// understands both the plain work key ("owner/repo#N") and the boundary's
// "owner/repo#<kind>/<target>" subject; a claim with no numeric target
// (branch pushes, issue/PR creation, repo claims, "!" keys) falls back to the
// repository scope.
func TargetWriterScope(c Claim) string {
if c.Type != ClaimTypeTask || len(c.Subject) <= len(c.Repo)+1 || c.Subject[len(c.Repo)] != '#' {
return c.Repo
}
rest := c.Subject[len(c.Repo)+1:]
if i := strings.IndexByte(rest, '/'); i >= 0 {
rest = rest[i+1:]
}
if !isDecimal(rest) {
return c.Repo
}
return c.Repo + "#" + rest
}

func isDecimal(s string) bool {
if s == "" {
return false
}
for i := 0; i < len(s); i++ {
if s[i] < '0' || s[i] > '9' {
return false
}
}
return true
}

// Entry is one durable claim record: the canonical claim, its holder, its
// monotonic epoch (per key, never reused — it survives Released and Waiting
// so a reacquisition is ALWAYS strictly higher), its lifecycle state, and its
Expand Down Expand Up @@ -84,6 +132,7 @@
type Ledger struct {
path string
maxWriters int
scope WriterScope
mu sync.Mutex
entries map[string]*Entry
}
Expand All @@ -103,7 +152,22 @@
if maxWritersPerRepo <= 0 {
maxWritersPerRepo = DefaultMaxWritersPerRepo
}
l := &Ledger{path: path, maxWriters: maxWritersPerRepo, entries: make(map[string]*Entry)}
return OpenLedgerScoped(path, maxWritersPerRepo, RepoWriterScope)
}

// OpenLedgerScoped is OpenLedger with an explicit writer scope: maxWriters
// slots are available per scope bucket rather than per repository. A nil
// scope selects RepoWriterScope; maxWriters <= 0 selects
// DefaultMaxWritersPerTarget for TargetWriterScope-style scoping and
// DefaultMaxWritersPerRepo otherwise (both are 1).
func OpenLedgerScoped(path string, maxWriters int, scope WriterScope) (*Ledger, error) {
if scope == nil {
scope = RepoWriterScope
}
if maxWriters <= 0 {
maxWriters = DefaultMaxWritersPerTarget
}
l := &Ledger{path: path, maxWriters: maxWriters, scope: scope, entries: make(map[string]*Entry)}
if err := l.reloadLocked(); err != nil {
return nil, err
}
Expand Down Expand Up @@ -179,12 +243,21 @@
return changed
}

// activeWritersLocked counts consumed mutation-writer slots in a repo:
// ONLY ActiveMutation consumes; Waiting and Released are structural zeros.
func (l *Ledger) activeWritersLocked(repo string) int {
func (l *Ledger) writerScope(c Claim) string {
if l.scope == nil {
return c.Repo
}
return l.scope(c)
}

// activeWritersLocked counts consumed mutation-writer slots in one writer
// scope (the repo under RepoWriterScope, repo#number under
// TargetWriterScope): ONLY ActiveMutation consumes; Waiting and Released are
// structural zeros.
func (l *Ledger) activeWritersLocked(scope string) int {
n := 0
for _, e := range l.entries {
if e.State == StateActiveMutation && e.Claim.Repo == repo {
if e.State == StateActiveMutation && l.writerScope(e.Claim) == scope {
n++
}
}
Expand All @@ -198,9 +271,10 @@
// 2. Any actively held overlapping claim refuses acquisition (ErrClaimHeld):
// exactly one owner per overlap set. The refused acquirer consumes no
// capacity — it simply does not acquire.
// 3. A repo with all writer slots consumed refuses (ErrNoCapacity), unless
// the acquisition is the reacquisition of this exact key (its own prior
// hold is the one being replaced).
// 3. A writer scope (the repo by default, repo#number under
// TargetWriterScope) with all writer slots consumed refuses
// (ErrNoCapacity), unless the acquisition is the reacquisition of this
// exact key (its own prior hold is the one being replaced).
// 4. Otherwise the entry moves to ActiveMutation at epoch prev+1, persisted
// before the grant is returned (durable-before-authoritative).
func (l *Ledger) Acquire(c Claim, holder string, ttl time.Duration, now time.Time) (Entry, error) {
Expand Down Expand Up @@ -236,8 +310,11 @@
key, otherKey, e.Holder, e.Epoch, ErrClaimHeld)
}
}
if l.activeWritersLocked(c.Repo) >= l.maxWriters {
return Entry{}, fmt.Errorf("repo %s writers exhausted (max %d): %w", c.Repo, l.maxWriters, ErrNoCapacity)
if scope := l.writerScope(c); l.activeWritersLocked(scope) >= l.maxWriters {
if scope == c.Repo {
return Entry{}, fmt.Errorf("repo %s writers exhausted (max %d): %w", c.Repo, l.maxWriters, ErrNoCapacity)
}
return Entry{}, fmt.Errorf("target %s writers exhausted (max %d): %w", scope, l.maxWriters, ErrNoCapacity)
}

var prevEpoch uint64
Expand Down
Loading
Loading