diff --git a/changelog.d/fixed-14645.md b/changelog.d/fixed-14645.md new file mode 100644 index 0000000000..9304b0c75a --- /dev/null +++ b/changelog.d/fixed-14645.md @@ -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)). diff --git a/src/cmd/hive/main.go b/src/cmd/hive/main.go index 31cbbe4abb..b9c76792bb 100644 --- a/src/cmd/hive/main.go +++ b/src/cmd/hive/main.go @@ -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 diff --git a/src/pkg/config/convergence.go b/src/pkg/config/convergence.go index 24fdac114c..2e5af84953 100644 --- a/src/pkg/config/convergence.go +++ b/src/pkg/config/convergence.go @@ -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 ( diff --git a/src/pkg/config/convergence_test.go b/src/pkg/config/convergence_test.go index 9eacb890a9..a202dcb890 100644 --- a/src/pkg/config/convergence_test.go +++ b/src/pkg/config/convergence_test.go @@ -1,6 +1,10 @@ package config -import "testing" +import ( + "testing" + + "gopkg.in/yaml.v3" +) // ── kubestellar/hive#3845 convergence feature toggle ─────────────────────────── // @@ -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) + } +} diff --git a/src/pkg/convergence/mutation/boundary.go b/src/pkg/convergence/mutation/boundary.go index 26b6fca2b1..ea1e2c6e04 100644 --- a/src/pkg/convergence/mutation/boundary.go +++ b/src/pkg/convergence/mutation/boundary.go @@ -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 { diff --git a/src/pkg/convergence/mutation/ledger.go b/src/pkg/convergence/mutation/ledger.go index 5b873f4eec..50488f5660 100644 --- a/src/pkg/convergence/mutation/ledger.go +++ b/src/pkg/convergence/mutation/ledger.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "sort" + "strings" "sync" "syscall" "time" @@ -33,7 +34,8 @@ var ( // 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 @@ -54,6 +56,52 @@ const ledgerFormatVersion = 1 // 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#/" 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 @@ -84,6 +132,7 @@ type persistedLedger struct { type Ledger struct { path string maxWriters int + scope WriterScope mu sync.Mutex entries map[string]*Entry } @@ -103,7 +152,22 @@ func OpenLedger(path string, maxWritersPerRepo int) (*Ledger, error) { 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 } @@ -179,12 +243,21 @@ func (l *Ledger) reconcileExpiredLocked(now time.Time) bool { 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++ } } @@ -198,9 +271,10 @@ func (l *Ledger) activeWritersLocked(repo string) int { // 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) { @@ -236,8 +310,11 @@ func (l *Ledger) Acquire(c Claim, holder string, ttl time.Duration, now time.Tim 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 diff --git a/src/pkg/convergence/mutation/ledger_scope_test.go b/src/pkg/convergence/mutation/ledger_scope_test.go new file mode 100644 index 0000000000..f7712c075d --- /dev/null +++ b/src/pkg/convergence/mutation/ledger_scope_test.go @@ -0,0 +1,153 @@ +package mutation + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/hivecommons/hive/pkg/effects" +) + +func openTargetLedger(t *testing.T, maxWriters int) *Ledger { + t.Helper() + l, err := OpenLedgerScoped(filepath.Join(t.TempDir(), "claims.json"), maxWriters, TargetWriterScope) + if err != nil { + t.Fatalf("OpenLedgerScoped: %v", err) + } + return l +} + +func TestTargetWriterScope(t *testing.T) { + cases := []struct { + claim Claim + want string + }{ + {TaskClaim("acme/widgets", "acme/widgets#label_mutation/7"), "acme/widgets#7"}, + {TaskClaim("acme/widgets", "acme/widgets#review_submit/7"), "acme/widgets#7"}, + {TaskClaim("acme/widgets", "acme/widgets#7"), "acme/widgets#7"}, + {TaskClaim("acme/widgets", "acme/widgets#branch_push/feat/123"), "acme/widgets"}, + {TaskClaim("acme/widgets", "acme/widgets#issue_create/abc123"), "acme/widgets"}, + {TaskClaim("acme/widgets", "acme/widgets#pull_request_create/feature"), "acme/widgets"}, + {TaskClaim("acme/widgets", "acme/widgets!KEY"), "acme/widgets"}, + {RepoClaim("acme/widgets"), "acme/widgets"}, + } + for _, tc := range cases { + if got := TargetWriterScope(tc.claim); got != tc.want { + t.Errorf("TargetWriterScope(%s) = %q, want %q", tc.claim.Key(), got, tc.want) + } + } +} + +// #14645: two writers on different PRs in the same repo both acquire; the +// per-repo ledger (OpenLedger) still refuses the second. +func TestLedgerTargetScope_DifferentTargetsSameRepoBothAcquire(t *testing.T) { + now := time.Unix(100, 0) + a := TaskClaim("acme/widgets", "acme/widgets#review_submit/7") + b := TaskClaim("acme/widgets", "acme/widgets#label_mutation/9") + + l := openTargetLedger(t, 0) + if _, err := l.Acquire(a, "reviewer", ttl, now); err != nil { + t.Fatalf("first acquire: %v", err) + } + if _, err := l.Acquire(b, "labeler", ttl, now); err != nil { + t.Fatalf("a writer on a different PR in the same repo must acquire: %v", err) + } + + repo, err := OpenLedger(filepath.Join(t.TempDir(), "claims.json"), 0) + if err != nil { + t.Fatalf("OpenLedger: %v", err) + } + if _, err := repo.Acquire(a, "reviewer", ttl, now); err != nil { + t.Fatalf("first acquire: %v", err) + } + if _, err := repo.Acquire(b, "labeler", ttl, now); !errors.Is(err, ErrNoCapacity) { + t.Fatalf("the per-repo ledger keeps its single repo slot: %v", err) + } +} + +func TestLedgerTargetScope_SameTargetSecondWriterRefused(t *testing.T) { + now := time.Unix(100, 0) + l := openTargetLedger(t, 0) + if _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#review_submit/7"), "reviewer", ttl, now); err != nil { + t.Fatalf("first acquire: %v", err) + } + _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#label_mutation/7"), "labeler", ttl, now) + if !errors.Is(err, ErrNoCapacity) { + t.Fatalf("second writer on the same repo#number must be refused: %v", err) + } + if !strings.Contains(err.Error(), "target acme/widgets#7 writers exhausted (max 1)") { + t.Fatalf("error must name the exhausted target: %v", err) + } +} + +func TestLedgerTargetScope_UnnumberedFallsBackToRepo(t *testing.T) { + now := time.Unix(100, 0) + l := openTargetLedger(t, 0) + if _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#branch_push/feature"), "pusher", ttl, now); err != nil { + t.Fatalf("first acquire: %v", err) + } + _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#pull_request_create/feature"), "opener", ttl, now) + if !errors.Is(err, ErrNoCapacity) || !strings.Contains(err.Error(), "repo acme/widgets writers exhausted (max 1)") { + t.Fatalf("unnumbered mutations share the repo slot: %v", err) + } + if _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#label_mutation/7"), "labeler", ttl, now); err != nil { + t.Fatalf("a numbered target does not queue behind the repo pool: %v", err) + } +} + +func TestLedgerTargetScope_MaxWritersHonoured(t *testing.T) { + now := time.Unix(100, 0) + l := openTargetLedger(t, 2) + for _, kind := range []string{"review_submit", "label_mutation"} { + if _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#"+kind+"/7"), kind, ttl, now); err != nil { + t.Fatalf("acquire %s within max 2: %v", kind, err) + } + } + _, err := l.Acquire(TaskClaim("acme/widgets", "acme/widgets#issue_comment/7"), "commenter", ttl, now) + if !errors.Is(err, ErrNoCapacity) || !strings.Contains(err.Error(), "(max 2)") { + t.Fatalf("third writer on the target must be refused at max 2: %v", err) + } +} + +// #14645 ask 3: label writes on two different issues are admitted at enforce +// while a review on a third PR in the same repo is in flight. +func TestBoundaryTargetScope_LabelsNotBlockedByInFlightReview(t *testing.T) { + journal, err := OpenJournal(filepath.Join(t.TempDir(), "journal.json")) + if err != nil { + t.Fatal(err) + } + ledger := openTargetLedger(t, 0) + now := time.Unix(100, 0) + b := &Boundary{Executor: Executor{Ledger: ledger, Journal: journal, Mode: "enforce", Now: func() time.Time { return now }}, Holder: "test"} + + review, err := ledger.Acquire(TaskClaim("acme/widgets", "acme/widgets#review_submit/3"), "reviewer", time.Hour, now) + if err != nil { + t.Fatal(err) + } + for _, target := range []string{"5", "6"} { + called := false + _, err := b.Execute(context.Background(), effects.Claim{Repo: "acme/widgets", Kind: effects.KindLabelMutation, Target: target}, + func(context.Context) (effects.Result, error) { + called = true + return effects.Result{Provenance: "labeled"}, nil + }) + if err != nil || !called { + t.Fatalf("label on #%s behind in-flight review on #3: called=%v err=%v", target, called, err) + } + } + called := false + _, err = b.Execute(context.Background(), effects.Claim{Repo: "acme/widgets", Kind: effects.KindLabelMutation, Target: "3"}, + func(context.Context) (effects.Result, error) { + called = true + return effects.Result{}, nil + }) + if !errors.Is(err, effects.ErrDenied) || called { + t.Fatalf("label on the reviewed PR itself must still wait: called=%v err=%v", called, err) + } + if _, err := ledger.Release(review.Claim.Key(), review.Epoch, now); err != nil { + t.Fatal(err) + } +} diff --git a/src/pkg/dashboard/settings_parity_allowlist.txt b/src/pkg/dashboard/settings_parity_allowlist.txt index 803a3107e7..270c821aa1 100644 --- a/src/pkg/dashboard/settings_parity_allowlist.txt +++ b/src/pkg/dashboard/settings_parity_allowlist.txt @@ -159,6 +159,7 @@ classification.review_bots.resolve_after_fix # TODO #11854 compliance.frameworks[] # TODO #11854 contribute.token_idle_max # TODO #11854 contribute.token_max_age # TODO #11854 +convergence.mutation.max_writers_per_target # boundary capacity tuning; yaml-only (#14645) deployment.runtime # set via HIVE_DEPLOYMENT_RUNTIME env var or auto-detected, not a Settings control (#14072) duplicate_sweep.bot_authors[] # TODO #11854 evidence.complexity.enabled # TODO #11854