Add RepositoryOperationGuard circuit breaker by harshitakaushik-dev · Pull Request #22523 · opensearch-project/OpenSearch

harshitakaushik-dev · 2026-07-21T09:12:33Z

Description

Adds RepositoryOperationGuard, a per-repository circuit breaker that bounds the number of outstanding cluster-manager-side repository operations. This guards against a degraded repository parking enough worker threads (via the upcoming operation-level timeout) to drain the small, shared snapshot thread pool and starve other repositories.

This class is registered nowhere and consumed by no one yet. A follow-up PR wires it into the finalize/delete dispatch paths.

Method	Purpose
`tryAcquire(repo)`	Fails fast with `RepositoryException` once a repository is at its limit, without dispatching
`release(repo)`	Releases a permit; must be called exactly once per successful acquire
`setMaxOutstandingOps(int)`	Updates the limit dynamically

Testing

RepositoryOperationGuardTests: fail-fast at limit, no permit taken on rejection.
Release balances acquire; dynamic limit updates apply to subsequent acquisitions.
Non-positive limits rejected at construction and via dynamic update.
Exactly-once release across response/failure/timeout, verified against the real ActionListener.runAfter + ListenerTimeouts stack, including a timeout followed by a late orphaned
completion.
Independent per-repository tracking.
Deterministic timing via DeterministicTaskQueue; no Thread.sleep.
No IT: this class has no reachable call site, so there is no observable behavior to exercise end-to-end yet. That coverage lands in the follow-up PR that wires this guard in.

Related Issues

Check List

Functionality includes testing.
API changes companion pull request created, if applicable.
Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check
here.

github-actions · 2026-07-21T09:13:31Z

PR Reviewer Guide 🔍

(Review updated until commit `d7762d1`)

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 No multiple PR themes
⚡ No major issues detected

github-actions · 2026-07-21T09:13:50Z

PR Code Suggestions ✨

Latest suggestions up to d7762d1
Explore these optional code suggestions:

Category	Suggestion	Impact
General	Avoid throwing inside compute remapping function Throwing from inside `ConcurrentHashMap.compute` propagates the exception, but the map's internal behavior on exception-in-remapping is that the mapping is left unchanged — which is the intent here. However, wrapping the throw inside `compute` can obscure stack traces and, more importantly, `RepositoryException` is a runtime exception that may be wrapped or logged unexpectedly by ConcurrentHashMap internals in some JDKs. Consider computing the decision outside `compute` or capturing the rejection and throwing after `compute` returns to make control flow clearer and avoid throwing inside a remapping function. server/src/main/java/org/opensearch/snapshots/RepositoryOperationGuard.java [45-54] public void tryAcquire(String repository) { final int limit = maxOutstandingOps; - outstandingOpsByRepository.compute(repository, (repo, current) -> { + final Integer result = outstandingOpsByRepository.compute(repository, (repo, current) -> { final int count = (current == null) ? 0 : current; if (count >= limit) { - throw new RepositoryException(repo, "[" + count + "] outstanding repository operations at the limit of [" + limit + "]"); + return current; // reject: leave count unchanged } return count + 1; }); + if (result != null && result > limit) { + // unreachable + } + if (getOutstandingOps(repository) >= limit && (result == null \|\| result >= limit && result == (result))) { + // Re-check to detect rejection; better: use a sentinel/AtomicBoolean captured in the lambda. + throw new RepositoryException(repository, "[" + result + "] outstanding repository operations at the limit of [" + limit + "]"); + } } Suggestion importance[1-10]: 2 __ Why: Throwing from within `ConcurrentHashMap.compute` is safe and idiomatic; the mapping is left unchanged on exception. The proposed `improved_code` is convoluted, has unreachable/incorrect logic, and introduces a race by re-reading state outside the atomic operation, making it worse than the original.	Low

Previous suggestions

Suggestions up to commit 928cc44

Category	Suggestion	Impact
Possible issue	Prevent counter from going negative on over-release If `release()` is called without a matching `tryAcquire()` (or over-released), the counter can go negative in production, permanently corrupting capacity for that repository. Guard the decrement with a CAS loop so the counter never drops below zero, keeping the assertion for tests but ensuring safe behavior in prod. server/src/main/java/org/opensearch/snapshots/RepositoryOperationGuard.java [63-72] public void release(String repository) { final AtomicInteger outstanding = outstandingOpsByRepository.get(repository); if (outstanding == null) { - // Caller bug: release() without a matching tryAcquire(). Throws AssertionError in tests (-ea), no-ops in prod. assert false : "release() called for repository [" + repository + "] with no outstanding operations tracked"; return; } - final int updated = outstanding.decrementAndGet(); - assert updated >= 0 : "release() over-released for repository [" + repository + "]"; + while (true) { + final int current = outstanding.get(); + if (current <= 0) { + assert false : "release() over-released for repository [" + repository + "]"; + return; + } + if (outstanding.compareAndSet(current, current - 1)) { + return; + } + } } Suggestion importance[1-10]: 6 __ Why: Valid defensive concern: in production without assertions, an over-release could drive the counter negative and effectively grant extra capacity. The CAS-guarded decrement is a reasonable safeguard, though the underlying bug is a caller error.	Low
General	Address unbounded map growth for repositories `outstandingOpsByRepository` grows unboundedly as new repository names appear and is never cleaned up, which is a memory leak for workloads with churning/renamed repositories. Consider removing entries when the counter reaches zero (with care to avoid races), or providing a `remove(String repository)` API called on repository deletion. server/src/main/java/org/opensearch/snapshots/RepositoryOperationGuard.java [46-61] +public void tryAcquire(String repository) { + final AtomicInteger outstanding = outstandingOpsByRepository.computeIfAbsent(repository, ignored -> new AtomicInteger(0)); + final int limit = maxOutstandingOps; + while (true) { + final int current = outstanding.get(); + if (current >= limit) { + throw new RepositoryException( + repository, + "circuit open — repository unreachable: [" + current + "] outstanding operations at limit [" + limit + "]" + ); + } + if (outstanding.compareAndSet(current, current + 1)) { + return; + } + } +} - Suggestion importance[1-10]: 5 __ Why: Legitimate observation about a potential memory leak with churning repository names, but the `improved_code` is identical to the `existing_code`, so no actual fix is provided.	Low

shourya035 · 2026-07-21T09:57:56Z

+            if (current >= limit) {
+                throw new RepositoryException(
+                    repository,
+                    "circuit open — repository unreachable: [" + current + "] outstanding operations at limit [" + limit + "]"


Let's not use circuit open in the exception message here. This is an implementation detail that we are using a circuit breaker logic.

shourya035 · 2026-07-21T10:06:59Z

+ * @opensearch.internal
+ */
+public final class RepositoryOperationGuard {
+    private final Map<String, AtomicInteger> outstandingOpsByRepository = new ConcurrentHashMap<>();


We should also have a cleanup method to clear off cached repository keys in this map once the repository is removed from the cluster. Or else this would be a point of memory leak

Makes sense. Going with cleanup-on-zero: release() will remove a repo's map entry once its count hits 0, done atomically via ConcurrentHashMap.compute (so it can't race with a concurrent tryAcquire on the same key). This bounds the map to repos with currently in-flight ops, a removed repo's entry drains out on its own once its last op completes, no separate hook needed.

github-actions · 2026-07-21T10:07:07Z

✅ Gradle check result for 928cc44: SUCCESS

shourya035 · 2026-07-21T10:07:56Z

+            assert false : "release() called for repository [" + repository + "] with no outstanding operations tracked";
+            return;
+        }
+        final int updated = outstanding.decrementAndGet();


This needs a guard to prevent this from going into negative

codecov · 2026-07-21T10:07:58Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.47%. Comparing base (0213338) to head (d7762d1).
⚠️ Report is 7 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #22523      +/-   ##
============================================
- Coverage     73.48%   73.47%   -0.01%     
+ Complexity    76460    76433      -27     
============================================
  Files          6104     6104              
  Lines        346582   346582              
  Branches      49879    49879              
============================================
- Hits         254675   254648      -27     
+ Misses        71664    71630      -34     
- Partials      20243    20304      +61

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

alchemist51 · 2026-07-21T11:51:04Z

+     * Acquires a permit for {@code repository}, or throws {@link RepositoryException} if the circuit is open. On
+     * success the caller must call {@link #release(String)} exactly once for this acquisition.
+     */
+    public void tryAcquire(String repository) {


Does this needs to be synchronized?

No synchronization needed here. The CAS (compareAndSet) already makes this thread-safe. That said, I'm switching to ConcurrentHashMap.compute for the rewrite, which does the check-and-increment atomically under the map's per-key lock, so the thread-safety is structural instead of something you have to trace through CAS logic.

Can we add concurrency tests?

alchemist51 · 2026-07-21T11:54:03Z

+    public void tryAcquire(String repository) {
+        final AtomicInteger outstanding = outstandingOpsByRepository.computeIfAbsent(repository, ignored -> new AtomicInteger(0));
+        final int limit = maxOutstandingOps;
+        while (true) {


Why do we have a while loop? We are throwing anyways above the limit and for lower limit we are incrementing the value

Removed. Switched tryAcquire/release to ConcurrentHashMap.compute, so the loop isn't needed anymore.

alchemist51 · 2026-07-21T11:54:45Z

+                    "circuit open — repository unreachable: [" + current + "] outstanding operations at limit [" + limit + "]"
+                );
+            }
+            if (outstanding.compareAndSet(current, current + 1)) {


What if any other thread increased the value for outstanding while we reach here?

Should not be possible now, the check-and-increment happens inside the same compute call, which is atomic per key.

Bounds outstanding cluster-manager-side repository operations per repository so a degraded store cannot exhaust the shared snapshot thread pool. Unwired for now; PR-F and PR-G will consume it. Signed-off-by: Harshita Kaushik <harycash@amazon.com>

github-actions · 2026-07-22T05:25:02Z

Persistent review updated to latest commit d7762d1

github-actions · 2026-07-22T06:21:17Z

✅ Gradle check result for d7762d1: SUCCESS

alchemist51 · 2026-07-22T07:00:31Z

Can we try to see if we can use the GatedCloseable<T> or AutoCloseableRefCounted to handle the release and close of repo references cleanly? @shourya035 @harshitakaushik-dev

alchemist51 · 2026-07-22T06:46:39Z

+     * Acquires a permit for {@code repository}, or throws {@link RepositoryException} if the circuit is open. On
+     * success the caller must call {@link #release(String)} exactly once for this acquisition.
+     */
+    public void tryAcquire(String repository) {


Can we add concurrency tests?

alchemist51 · 2026-07-22T06:47:23Z

+    public void release(String repository) {
+        outstandingOpsByRepository.compute(repository, (repo, current) -> {
+            if (current == null || current <= 0) {
+                assert false : "release() with no outstanding operations tracked for repository [" + repo + "]";


Can we print a warn message here? Ideally this should not happen, similarly above

alchemist51 · 2026-07-22T06:52:46Z

+                return null;
+            }
+            final int updated = current - 1;
+            return updated == 0 ? null : updated;


Will this remove the "repo" key in case the updated has become zero?

Yes. Returning null from the compute remapping function removes the mapping (per ConcurrentHashMap.compute's contract). So the entry is dropped the moment the count reaches 0. That's the cleanup-on-zero. Have verified by inspecting the map directly.

harshitakaushik-dev · 2026-07-22T08:04:49Z

Can we try to see if we can use the GatedCloseable<T> or AutoCloseableRefCounted to handle the release and close of repo references cleanly? @shourya035 @harshitakaushik-dev

Checked on it, GatedCloseable<T> fits: its OneWayGate guarantees release() runs exactly once, so PR-F/PR-G can't leak a permit by forgetting to release on an exception path. AutoCloseableRefCounted doesn't fit here, it adapts a RefCounted object, and we're tracking a per-repo count in a map, not a ref-counted object.

One cost worth flagging: GatedCloseable implements Closeable, so close()/release() is throws IOException on the signature (even though our cleanup never actually throws). Every try-with-resources call site in PR-F/PR-G will need to declare or catch that, dead code for an exception that can't happen, but it's how the existing utility is shaped, and other OpenSearch code (Engine, InternalEngine, etc.) already lives with the same tax. Let me know if that seems good to you, unless you'd rather keep it simple @shourya035

shourya035 · 2026-07-22T09:39:58Z

Can we try to see if we can use the GatedCloseable<T> or AutoCloseableRefCounted to handle the release and close of repo references cleanly? @shourya035 @harshitakaushik-dev

Checked on it, GatedCloseable<T> fits: its OneWayGate guarantees release() runs exactly once, so PR-F/PR-G can't leak a permit by forgetting to release on an exception path. AutoCloseableRefCounted doesn't fit here, it adapts a RefCounted object, and we're tracking a per-repo count in a map, not a ref-counted object.

One cost worth flagging: GatedCloseable implements Closeable, so close()/release() is throws IOException on the signature (even though our cleanup never actually throws). Every try-with-resources call site in PR-F/PR-G will need to declare or catch that, dead code for an exception that can't happen, but it's how the existing utility is shaped, and other OpenSearch code (Engine, InternalEngine, etc.) already lives with the same tax. Let me know if that seems good to you, unless you'd rather keep it simple @shourya035

What is PR-F/PR-G? I didn't get this.

It should be fine to just add the throws statement to our method signature, but yes GatedCloseable help us with ensuring that the acquire/release cycle runs at-most once, which is what we need.

harshitakaushik-dev requested a review from a team as a code owner July 21, 2026 09:12

shourya035 reviewed Jul 21, 2026

View reviewed changes

alchemist51 requested changes Jul 21, 2026

View reviewed changes

harshitakaushik-dev force-pushed the fix/snapshot-resilience-pr-d-circuit-breaker branch from 928cc44 to d7762d1 Compare July 22, 2026 05:24

alchemist51 requested changes Jul 22, 2026

View reviewed changes

harshitakaushik-dev requested a review from shourya035 July 22, 2026 08:36

Uh oh!

Conversation

harshitakaushik-dev commented Jul 21, 2026

Description

Testing

Related Issues

Check List

Uh oh!

github-actions Bot commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Reviewer Guide 🔍

(Review updated until commit d7762d1)

Uh oh!

github-actions Bot commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Code Suggestions ✨

Previous suggestions

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 21, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

harshitakaushik-dev Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 22, 2026

Uh oh!

github-actions Bot commented Jul 22, 2026

Uh oh!

alchemist51 commented Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

harshitakaushik-dev commented Jul 22, 2026

Uh oh!

shourya035 commented Jul 22, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

github-actions Bot commented Jul 21, 2026 •

edited

Loading

(Review updated until commit `d7762d1`)

github-actions Bot commented Jul 21, 2026 •

edited

Loading

codecov Bot commented Jul 21, 2026 •

edited

Loading

harshitakaushik-dev Jul 22, 2026 •

edited

Loading

alchemist51 commented Jul 22, 2026 •

edited

Loading