From 99bf56f4022ddbe289c6f09a370227f809733415 Mon Sep 17 00:00:00 2001
From: ShivamPaliwal1 <shivam.paliwal@couchbase.com>
Date: Thu, 6 Aug 2026 21:24:00 +0530
Subject: [PATCH] Bound DocOps bulk operations with a timeout on the aggregate
 block()

Per-op timeouts on insert/upsert/get/delete/replace/touch only guarantee a
single request terminates; if the shared reactor scheduler wedges under
heavy backpressure (e.g. concurrent disk-full workloads), that signal can
fail to reach the blocked caller and .block() hangs the JVM indefinitely.
Adding an explicit .timeout() on the aggregate collectList() gives each
bulk call its own backstop instead of relying solely on an external
stall-watchdog to kill the process.
---
 src/main/java/couchbase/sdk/DocOps.java | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/src/main/java/couchbase/sdk/DocOps.java b/src/main/java/couchbase/sdk/DocOps.java
index 7e18c89..40bcb31 100644
--- a/src/main/java/couchbase/sdk/DocOps.java
+++ b/src/main/java/couchbase/sdk/DocOps.java
@@ -32,6 +32,14 @@
 
 public class DocOps {
 
+    // Backstop for the aggregate .block() below. Per-op timeouts (set on the passed-in
+    // *Options) bound a single request, but if the shared reactor scheduler/event-loop
+    // ever wedges under heavy backpressure (e.g. concurrent disk-full workloads), a
+    // per-op timeout firing internally does not guarantee its signal reaches the blocked
+    // caller. Without this, .block() can hang the calling thread indefinitely with no
+    // way for the JVM itself to recover.
+    private static final Duration BULK_OP_TIMEOUT = Duration.ofSeconds(120);
+
     public List<Result> bulkInsert(Collection collection, List<Tuple2<String, Object>> documents, InsertOptions insertOptions) {
         ReactiveCollection reactiveCollection = collection.reactive();
 
@@ -49,6 +57,7 @@ public List<Result> bulkInsert(Collection collection, List<Tuple2<String, Object
                           .onErrorResume(error -> Mono.just(new Result(k, v, error, false)));
                 }, concurrency)
                 .collectList()
+                .timeout(BULK_OP_TIMEOUT)
                 .block();
       }
 
@@ -70,6 +79,7 @@ public List<Result> bulkUpsert(Collection collection, List<Tuple2<String, Object
                           .onErrorResume(error -> Mono.just(new Result(k, v, error, false)));
                 }, concurrency)
                 .collectList()
+                .timeout(BULK_OP_TIMEOUT)
                 .block();
     }
 
@@ -92,7 +102,7 @@ public Mono<Tuple2<String, Object>> apply(Throwable error) {
                                     }
                                 });
                     }
-                }, concurrency).collectList().block();
+                }, concurrency).collectList().timeout(BULK_OP_TIMEOUT).block();
         return returnValue;
     }
 
@@ -110,6 +120,7 @@ public List<Result> bulkDelete(Collection collection, List<String> keys, RemoveO
                           .onErrorResume(error -> Mono.just(new Result(key, null, error, false)));
                 }, concurrency)
                 .collectList()
+                .timeout(BULK_OP_TIMEOUT)
                 .block();
     }
 
@@ -144,7 +155,7 @@ public Mono<ConcurrentHashMap<String, Object>> apply(Throwable error) {
                                     }
                                 });
                     }
-                }, concurrency).collectList().block();
+                }, concurrency).collectList().timeout(BULK_OP_TIMEOUT).block();
         return returnValue;
     }
 
@@ -175,7 +186,7 @@ public Mono<ConcurrentHashMap<String, Object>> apply(Throwable error) {
                             }
                                 }).defaultIfEmpty(returnValue);
                     }
-                }, concurrency).collectList().block();
+                }, concurrency).collectList().timeout(BULK_OP_TIMEOUT).block();
         return returnValue;
     }