From 3e6f2769980694f246b7eb952f0029e5ff0da8e9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 6 Jun 2026 04:51:14 +0000 Subject: [PATCH] Prefer local delivery for service RPC over the clustered event bus Service RPCs travel over the Vert.x clustered event bus addressed to the service's base resource. The point-to-point selector round-robins sends across every node registered for that address, so when a node hosts a service locally and the same service is also registered elsewhere, ~half of that node's own calls get shipped to a remote node for no reason. Track the addresses this node hosts a local handler for in a reference counted map, populated where the consumer's handler is attached in _listen and cleared (once) when it is unregistered, so membership means "a local consumer exists at this address on this node". On the send path, set DeliveryOptions.setLocalOnly(true) for point-to-point service sends whose base resource is locally hosted, which bypasses the cluster selector and delivers to the local handler. It is gated on the service scheme and only fires when a local handler exists, so it never misroutes and never trips NO_HANDLERS. The per-message check is a lock free containsKey plus a scheme check, no cluster or registry query. Works on Vert.x 4 and 5 (localOnly is the version-agnostic lever after Vert.x 5 removed the pluggable NodeSelector SPI). https://claude.ai/code/session_01QrcYMiSzEbVk9FrukdyS4U --- .../api/event/DefaultEventBusService.java | 61 +++++++- ...faultEventBusServiceLocalDeliveryTest.java | 144 ++++++++++++++++++ 2 files changed, 198 insertions(+), 7 deletions(-) create mode 100644 continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java diff --git a/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java b/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java index 1230cd2a..bb3a1a72 100644 --- a/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java +++ b/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java @@ -56,6 +56,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; /** * Default implementation of {@link EventBusService} using the vertx {@link io.vertx.core.eventbus.EventBus} as a backend @@ -76,6 +78,15 @@ public class DefaultEventBusService implements EventBusService { private IgniteCache> subscriptionsCache; @Autowired private Vertx vertx; + /** + * Reference counted set of event bus addresses this node hosts a local handler for. + * An entry is added by {@link #_listen} when a consumer's handler is attached on this node and removed + * when that consumer is unregistered, so membership means "a local consumer exists at this address". + * This is local by construction: {@link #_listen} only ever registers consumers on this node, other nodes' + * registrations live solely in the cluster's subscription registry. The send path uses this to prefer + * local delivery for service RPCs and avoid an unnecessary cluster hop. + */ + private final Map localListenerCounts = new ConcurrentHashMap<>(); @PostConstruct public void init(){ @@ -111,7 +122,7 @@ public Mono>> listenWithAck(String cri) { return Mono.create(sink -> { final MessageConsumer consumer = vertx.eventBus().consumer(cri); - final ConnectableFlux> flux = _listen(null, consumer).publish(); + final ConnectableFlux> flux = _listen(cri, consumer).publish(); consumer.completion().onComplete(ar ->{ if(ar.succeeded()){ sink.success(flux); @@ -185,8 +196,9 @@ public Flux monitorListenerStatus(String cri) { @Override public void send(Event event) { - DeliveryOptions deliveryOptions = createDeliveryOptions(event); - vertx.eventBus().send(event.cri().baseResource(), + String baseResource = event.cri().baseResource(); + DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource); + vertx.eventBus().send(baseResource, event.data(), deliveryOptions); } @@ -195,10 +207,11 @@ public void send(Event event) { public Mono sendWithAck(Event event) { Validate.notNull(event, "Event must not be null"); return Mono.create(sink -> { - DeliveryOptions deliveryOptions = createDeliveryOptions(event); + String baseResource = event.cri().baseResource(); + DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource); // We expect that a response will be sent upon receipt. This will happen automatically if the listener is created with this class. vertx.eventBus() - .request(event.cri().baseResource(), + .request(baseResource, event.data(), deliveryOptions) .onComplete(reply -> { @@ -220,8 +233,20 @@ private Flux> _listen(String cri, MessageConsumer vertxEve } Flux> ret = Flux.create(fluxSink -> { + // Record that this node now hosts a local handler for cri. Done here, where the consumer's handler is + // attached, and undone in onDispose, where the consumer is unregistered, so the count is coupled to the + // consumer's actual registration lifecycle on this node. + localListenerCounts.merge(cri, 1, Integer::sum); + final AtomicBoolean unregistered = new AtomicBoolean(false); + // Setup all required handlers that are needed prior to consuming messages - fluxSink.onDispose(consumer::unregister); + fluxSink.onDispose(() -> { + // Decrement exactly once on the first unregister, removing the entry when no local handlers remain. + if(unregistered.compareAndSet(false, true)){ + localListenerCounts.computeIfPresent(cri, (k, n) -> n == 1 ? null : n - 1); + } + consumer.unregister(); + }); // TODO: deal with back pressure properly.. ? //fluxSink.onRequest() @@ -245,7 +270,7 @@ private Flux> _listen(String cri, MessageConsumer vertxEve return ret; // ensure message delivery happens on vertx event loop, not sure but this by itself did not move the next above to the work loop } - private DeliveryOptions createDeliveryOptions(Event event){ + private DeliveryOptions createDeliveryOptions(Event event, String baseResource){ DeliveryOptions deliveryOptions = new DeliveryOptions(); deliveryOptions.setTracingPolicy(TracingPolicy.IGNORE); // fast path for MultiMapMetadataAdapter's @@ -257,7 +282,29 @@ private DeliveryOptions createDeliveryOptions(Event event){ } } deliveryOptions.addHeader(EventConstants.CRI_HEADER, event.cri().raw()); + + // Prefer local delivery when this node already hosts a handler for the target service address. + // In the clustered event bus the point-to-point selector round-robins sends across every node + // registered for the address, so without this ~half of a node's own service calls get shipped to a + // remote node. localOnly=true bypasses the cluster selector and delivers to the local handler. + // Only set when this node actually hosts a handler, otherwise localOnly would fail with NO_HANDLERS. + if(shouldPreferLocalDelivery(event, baseResource)){ + deliveryOptions.setLocalOnly(true); + } + return deliveryOptions; } + /** + * Determines if a send to the given {@code baseResource} should be delivered to a local handler only. + * True only for point-to-point {@link EventConstants#SERVICE_DESTINATION_SCHEME service} sends whose + * address is currently hosted by a local handler on this node. This is a lock free + * {@link Map#containsKey} plus a scheme check, no cluster or registry query. + * Package private for testing. + */ + boolean shouldPreferLocalDelivery(Event event, String baseResource){ + return EventConstants.SERVICE_DESTINATION_SCHEME.equals(event.cri().scheme()) + && localListenerCounts.containsKey(baseResource); + } + } diff --git a/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java b/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java new file mode 100644 index 00000000..325499c0 --- /dev/null +++ b/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java @@ -0,0 +1,144 @@ +/* + * + * Copyright 2008-2021 Kinotic and the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.kinotic.continuum.internal.core.api.event; + +import io.vertx.core.Vertx; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.kinotic.continuum.core.api.event.Event; +import org.springframework.test.util.ReflectionTestUtils; +import reactor.core.Disposable; + +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies the {@code localOnly} delivery predicate of {@link DefaultEventBusService}: when this node hosts a + * local handler for a service address, sends to that address should prefer local delivery, otherwise they should + * fall back to normal cluster routing. + * + * Uses a real (non clustered) {@link Vertx} so the consumer registration / unregistration lifecycle that drives + * the predicate is exercised end to end rather than mocked. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class DefaultEventBusServiceLocalDeliveryTest { + + private Vertx vertx; + private DefaultEventBusService eventBusService; + + @BeforeAll + public void startVertx() { + vertx = Vertx.vertx(); + } + + @AfterAll + public void stopVertx() { + if (vertx != null) { + vertx.close(); + } + } + + @BeforeEach + public void setUp() { + // a fresh service per test so the tracked local listener state starts empty + eventBusService = new DefaultEventBusService(); + ReflectionTestUtils.setField(eventBusService, "vertx", vertx); + eventBusService.init(); + } + + @AfterEach + public void tearDown() { + eventBusService = null; + } + + private static Event event(String cri) { + return Event.create(cri, "data".getBytes(StandardCharsets.UTF_8)); + } + + @Test + public void prefersLocalDeliveryForLocallyHostedServiceAddress() { + String baseResource = "srv://org.kinotic.continuum.tests.LocalDeliveryService"; + Disposable subscription = eventBusService.listen(baseResource).subscribe(); + try { + // a send targets the base resource of the request cri (scheme + optional scope + resource name) + Event request = event(baseResource + "/someMethod"); + assertThat(request.cri().baseResource()) + .as("the send address must equal the listened base resource") + .isEqualTo(baseResource); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a service address hosted by a local handler should prefer local delivery") + .isTrue(); + } finally { + subscription.dispose(); + } + } + + @Test + public void doesNotPreferLocalDeliveryForAddressNotHostedLocally() { + String hosted = "srv://org.kinotic.continuum.tests.HostedService"; + Disposable subscription = eventBusService.listen(hosted).subscribe(); + try { + Event request = event("srv://org.kinotic.continuum.tests.OtherService/someMethod"); + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a service address not hosted locally should fall back to cluster routing") + .isFalse(); + } finally { + subscription.dispose(); + } + } + + @Test + public void doesNotPreferLocalDeliveryForNonServiceScheme() { + // This node hosts a local handler at this exact address, but the non service scheme must still + // suppress the optimization so reply / stream / other sends are left untouched. + String streamAddress = "stream://org.kinotic.continuum.tests.SomeStream"; + Disposable subscription = eventBusService.listen(streamAddress).subscribe(); + try { + Event request = event(streamAddress + "/someMethod"); + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a non service scheme should not prefer local delivery even when hosted locally") + .isFalse(); + } finally { + subscription.dispose(); + } + } + + @Test + public void clearsAfterConsumerUnregistered() { + String baseResource = "srv://org.kinotic.continuum.tests.EphemeralService"; + Disposable subscription = eventBusService.listen(baseResource).subscribe(); + Event request = event(baseResource + "/someMethod"); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("local delivery should be preferred while the consumer is registered") + .isTrue(); + + // unregistering the only local handler must clear the preference so a later send falls back to the cluster + subscription.dispose(); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("local delivery preference should clear once the consumer is unregistered") + .isFalse(); + } +}