From 380213398d7dd10fba9854fdc5a800f6b8fe2abc Mon Sep 17 00:00:00 2001 From: vipzj Date: Sun, 23 Aug 2026 23:55:12 +0800 Subject: [PATCH 1/2] fix(dataplane): verify environment API key off the netty event loop EnvironmentKeyAuthFilter calls ControlPlaneClient.verifyEnvironmentKey from the reactive security filter chain (reactor-http-epoll thread). The WebClient chain inside ends with .block(), which Reactor rejects on NonBlocking threads, so verification always failed and every self-hosted environment worker request was rejected with 401: verifyEnvironmentKey failed for env_xxx: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-4 Run the blocking verification on a small dedicated thread pool instead. --- .../builder/control/ControlPlaneClient.java | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java index 719e905062..545fcd61fb 100644 --- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java +++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java @@ -22,6 +22,8 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +48,19 @@ @Service public class ControlPlaneClient { + /** + * Dedicated pool for {@link #verifyEnvironmentKey}: it is invoked from the reactive security + * filter chain, so the blocking verification HTTP call must not run on a NonBlocking thread. + */ + private final ExecutorService envKeyVerifyPool = + Executors.newFixedThreadPool( + 4, + r -> { + Thread t = new Thread(r, "env-key-verify"); + t.setDaemon(true); + return t; + }); + private static final Logger log = LoggerFactory.getLogger(ControlPlaneClient.class); private final WebClient webClient; @@ -329,6 +344,17 @@ public EnvironmentDto getEnvironment(String environmentId, String actingUserId) * so the auth filter can fall through without 5xx. */ public boolean verifyEnvironmentKey(String environmentId, String plaintextKey) { + try { + return envKeyVerifyPool + .submit(() -> doVerifyEnvironmentKey(environmentId, plaintextKey)) + .get(); + } catch (Exception ex) { + log.debug("verifyEnvironmentKey failed for {}: {}", environmentId, ex.getMessage()); + return false; + } + } + + private boolean doVerifyEnvironmentKey(String environmentId, String plaintextKey) { try { Map body = Map.of("key", plaintextKey); Map resp = From 73ce3f9c0122573f017fa9dd5d07fe5aa7d1bd00 Mon Sep 17 00:00:00 2001 From: vipzj Date: Sun, 23 Aug 2026 23:55:15 +0800 Subject: [PATCH 2/2] fix(dataplane): schedule self-hosted worker endpoints on boundedElastic SelfHostedWorkerController and WorkerEnvironmentController wrap blocking work (session resolve via ControlPlaneClient, work queue ops) in Mono.fromCallable without subscribeOn, so the callable runs on the subscribing parallel/event-loop thread and fails with: resolveSession failed for sess_xxx: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-1 The pending-tools endpoint returned 502 and work/poll failed intermittently, which made the self-hosted hands worker unusable. Add .subscribeOn(Schedulers.boundedElastic()) like the other data-plane controllers (DataSessionApiController) already do. --- .../builder/control/ControlPlaneClient.java | 26 ------- .../web/api/SelfHostedWorkerController.java | 67 +++++++++++-------- .../web/api/WorkerEnvironmentController.java | 51 +++++++------- 3 files changed, 66 insertions(+), 78 deletions(-) diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java index 545fcd61fb..719e905062 100644 --- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java +++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ControlPlaneClient.java @@ -22,8 +22,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,19 +46,6 @@ @Service public class ControlPlaneClient { - /** - * Dedicated pool for {@link #verifyEnvironmentKey}: it is invoked from the reactive security - * filter chain, so the blocking verification HTTP call must not run on a NonBlocking thread. - */ - private final ExecutorService envKeyVerifyPool = - Executors.newFixedThreadPool( - 4, - r -> { - Thread t = new Thread(r, "env-key-verify"); - t.setDaemon(true); - return t; - }); - private static final Logger log = LoggerFactory.getLogger(ControlPlaneClient.class); private final WebClient webClient; @@ -344,17 +329,6 @@ public EnvironmentDto getEnvironment(String environmentId, String actingUserId) * so the auth filter can fall through without 5xx. */ public boolean verifyEnvironmentKey(String environmentId, String plaintextKey) { - try { - return envKeyVerifyPool - .submit(() -> doVerifyEnvironmentKey(environmentId, plaintextKey)) - .get(); - } catch (Exception ex) { - log.debug("verifyEnvironmentKey failed for {}: {}", environmentId, ex.getMessage()); - return false; - } - } - - private boolean doVerifyEnvironmentKey(String environmentId, String plaintextKey) { try { Map body = Map.of("key", plaintextKey); Map resp = diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java index ec26a4205b..c1cb012b09 100644 --- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java +++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/SelfHostedWorkerController.java @@ -39,6 +39,7 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.server.ResponseStatusException; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; /** * Outbound-worker data plane for {@code self_hosted}: pending tool_use listing, tool_result @@ -74,10 +75,11 @@ public Mono>> pendingTools( @PathVariable("sessionId") String sessionId, Authentication auth) { return Mono.fromCallable( - () -> { - requireEnvironmentWorker(auth, environmentId, sessionId); - return pendingHandsToolService.listPending(sessionId); - }); + () -> { + requireEnvironmentWorker(auth, environmentId, sessionId); + return pendingHandsToolService.listPending(sessionId); + }) + .subscribeOn(Schedulers.boundedElastic()); } /** Posts one or more tool results and resumes the suspended turn. */ @@ -88,27 +90,33 @@ public Mono> toolResults( @RequestBody ToolResultsRequest body, Authentication auth) { return Mono.fromCallable( - () -> { - ManagedSessionDto session = - requireEnvironmentWorker(auth, environmentId, sessionId); - if (body == null || body.results() == null || body.results().isEmpty()) { - throw ApiException.invalidRequest( - "missing_results", "results is required", "results"); - } - List blocks = new ArrayList<>(); - List recorded = new ArrayList<>(); - for (Map payload : body.results()) { - ToolResultBlock block = SessionTurnRunner.toolResultFromPayload(payload); - blocks.add(block); - Map stored = new LinkedHashMap<>(payload); - stored.putIfAbsent("tool_use_id", block.getId()); - recorded.add( - eventLog.append( - sessionId, SessionEventTypes.USER_TOOL_RESULT, stored)); - } - turnRunner.resumeWithToolResults(session, blocks); - return recorded; - }); + () -> { + ManagedSessionDto session = + requireEnvironmentWorker(auth, environmentId, sessionId); + if (body == null + || body.results() == null + || body.results().isEmpty()) { + throw ApiException.invalidRequest( + "missing_results", "results is required", "results"); + } + List blocks = new ArrayList<>(); + List recorded = new ArrayList<>(); + for (Map payload : body.results()) { + ToolResultBlock block = + SessionTurnRunner.toolResultFromPayload(payload); + blocks.add(block); + Map stored = new LinkedHashMap<>(payload); + stored.putIfAbsent("tool_use_id", block.getId()); + recorded.add( + eventLog.append( + sessionId, + SessionEventTypes.USER_TOOL_RESULT, + stored)); + } + turnRunner.resumeWithToolResults(session, blocks); + return recorded; + }) + .subscribeOn(Schedulers.boundedElastic()); } /** Downloads the session agent's skills bundle for local staging on the worker. */ @@ -118,10 +126,11 @@ public Mono> skills( @PathVariable("sessionId") String sessionId, Authentication auth) { return Mono.fromCallable( - () -> { - requireEnvironmentWorker(auth, environmentId, sessionId); - return skillsBundleService.bundleForSession(sessionId); - }); + () -> { + requireEnvironmentWorker(auth, environmentId, sessionId); + return skillsBundleService.bundleForSession(sessionId); + }) + .subscribeOn(Schedulers.boundedElastic()); } private ManagedSessionDto requireEnvironmentWorker( diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java index dc37a0b9eb..76b45c67d4 100644 --- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java +++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/web/api/WorkerEnvironmentController.java @@ -37,6 +37,7 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.server.ResponseStatusException; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; /** * REST surface for out-of-process Environment Workers on {@code self_hosted} environments. @@ -76,19 +77,20 @@ public Mono> poll( @RequestParam("workerId") String workerId, @RequestParam(name = "timeoutMs", defaultValue = "25000") long timeoutMs) { return Mono.fromCallable( - () -> { - try { - Optional item = - workQueue.poll(environmentId, workerId, timeoutMs); - return item.map(this::withSessionMetadata) - .map(ResponseEntity::ok) - .orElseGet(() -> ResponseEntity.noContent().build()); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new ResponseStatusException( - HttpStatus.SERVICE_UNAVAILABLE, "Poll interrupted"); - } - }); + () -> { + try { + Optional item = + workQueue.poll(environmentId, workerId, timeoutMs); + return item.map(this::withSessionMetadata) + .map(ResponseEntity::ok) + .orElseGet(() -> ResponseEntity.noContent().build()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ResponseStatusException( + HttpStatus.SERVICE_UNAVAILABLE, "Poll interrupted"); + } + }) + .subscribeOn(Schedulers.boundedElastic()); } private EnvironmentWorkQueue.WorkItem withSessionMetadata(EnvironmentWorkQueue.WorkItem item) { @@ -109,7 +111,8 @@ public Mono> listWork( @RequestParam(value = "state", required = false) String state, Authentication auth) { requireUserAuth(auth); - return Mono.fromCallable(() -> workQueue.list(environmentId, state)); + return Mono.fromCallable(() -> workQueue.list(environmentId, state)) + .subscribeOn(Schedulers.boundedElastic()); } /** Returns per-status counts and oldest queued age for the environment. */ @@ -117,7 +120,8 @@ public Mono> listWork( public Mono workStats( @PathVariable("id") String environmentId, Authentication auth) { requireUserAuth(auth); - return Mono.fromCallable(() -> workQueue.stats(environmentId)); + return Mono.fromCallable(() -> workQueue.stats(environmentId)) + .subscribeOn(Schedulers.boundedElastic()); } /** Returns a single work item by id. */ @@ -128,14 +132,15 @@ public Mono getWork( Authentication auth) { requireUserAuth(auth); return Mono.fromCallable( - () -> - workQueue - .get(workId) - .orElseThrow( - () -> - new ResponseStatusException( - HttpStatus.NOT_FOUND, - "Unknown work item: " + workId))); + () -> + workQueue + .get(workId) + .orElseThrow( + () -> + new ResponseStatusException( + HttpStatus.NOT_FOUND, + "Unknown work item: " + workId))) + .subscribeOn(Schedulers.boundedElastic()); } /**