diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 75eb8be9..c92baabf 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -1,5 +1,4 @@ stages: - - ee_tests - package variables: @@ -35,19 +34,6 @@ cache: KUBERNETES_SERVICE_MEMORY_REQUEST: "2Gi" KUBERNETES_SERVICE_MEMORY_LIMIT: "10Gi" -# Trigger `Tarantool Java SDK` enterprise tests run -run_ee_tests: - stage: ee_tests - allow_failure: true - trigger: - project: tarantool/java/tarantool-java-sdk-ee-testing - strategy: depend - branch: master - variables: - MAIN_REPO_SHA: "${CI_COMMIT_SHA}" - MAIN_REPO_COMMIT_REF_NAME: "${CI_COMMIT_REF_NAME}" - PARENT_PIPELINE_ID: "${CI_PIPELINE_ID}" - run_ee_release: stage: package allow_failure: true diff --git a/CHANGELOG.md b/CHANGELOG.md index 658eb0f6..6f0002be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ - Support ISO 8601 duration parsing and formatting for `Interval` +### Bug fixes + +- Handle `ER_AUTH_REQUIRED` for watcher registration. Watchers (including the automatic + `box.shutdown` watcher created when `gracefulShutdown` is enabled) are now deferred when + the server rejects their registration before authentication and re-registered after + `authorize()` completes. Tarantool EE builds with option `security.disable_guest: true` + reject pre-authentication watcher registration with `ER_AUTH_REQUIRED`; since 1.7.0 that + watcher error failed the whole connect procedure and put the connection into an endless + reconnect loop. + ## [1.7.1] - 2026-08-31 ### Dependencies diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index c7d9b27a..b248c732 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -101,6 +101,7 @@ public class IProtoClientImpl implements IProtoClient { private final WatcherOptions watcherOpts; private CompletableFuture serverProtocolVersion; private CompletableFuture> serverFeatures; + private volatile boolean authorized; private Set clientFeaturesEnum; private List clientFeaturesList; private LongTaskTimer requestTimer; @@ -185,11 +186,10 @@ public CompletableFuture connect(InetSocketAddress address, long timeoutMs public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { if (gracefulShutdown) { - // it does not send watch message if connection is not connected, - // it sends immediately after successful connect watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } + authorized = false; serverProtocolVersion = new CompletableFuture<>(); serverFeatures = new CompletableFuture<>(); return connection @@ -227,7 +227,14 @@ public CompletableFuture authorize( promise.completeExceptionally(new ClientException("No greeting, connect firstly!")); return promise; } - return runRequest(new IProtoAuth(user, password, greeting.get().getSalt(), authType), opts); + // re-register watchers rejected before authentication + return runRequest(new IProtoAuth(user, password, greeting.get().getSalt(), authType), opts) + .thenApply( + response -> { + authorized = true; + updateWatchers(); + return response; + }); } @Override @@ -617,7 +624,7 @@ public CompletableFuture execute( @Override public CompletableFuture ping() { - return runRequest(new IProtoPing(), DEFAULT_REQUEST_OPTS); + return ping(DEFAULT_REQUEST_OPTS); } @Override @@ -807,7 +814,13 @@ private synchronized void updateWatchers() { long syncId = allocateSyncIds(1); fsm = new WatcherStateMachine( - watcherEntry.getKey(), syncId, watcher, connection, watcherOpts, timerService); + watcherEntry.getKey(), + syncId, + watcher, + connection, + watcherOpts, + timerService, + failed -> onWatchAuthRequired(watcher, failed)); watcher.setStateContext(fsm); watcher.setSyncId(syncId); fsmRegistry.put(syncId, fsm); @@ -831,6 +844,21 @@ private synchronized void updateWatchers() { } } + /** + * Resets a watcher rejected before authentication, so that it is re-registered after authorize() + * or immediately, if the client is already authorized. + */ + private synchronized void onWatchAuthRequired(Watcher watcher, WatcherStateMachine failed) { + fsmRegistry.remove(failed.getSyncId()); + // a newer registration attempt may already be active + if (watcher.getStateContext() == failed) { + watcher.setStateContext(null); + } + if (authorized) { + updateWatchers(); + } + } + protected long allocateSyncIds(int count) { // n + count < 0 is a check for overflow return syncIdSequence.updateAndGet(n -> (n + count) < 0 ? count : (n + count)); diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java index f4e6a986..566d64cc 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/fsm/WatcherStateMachine.java @@ -19,6 +19,7 @@ import io.tarantool.core.exceptions.ClientException; import io.tarantool.core.protocol.IProtoRequest; import io.tarantool.core.protocol.IProtoResponse; +import io.tarantool.core.protocol.requests.IProtoConstant; import io.tarantool.core.protocol.requests.IProtoWatch; public class WatcherStateMachine implements IProtoStateMachine { @@ -37,8 +38,12 @@ public class WatcherStateMachine implements IProtoStateMachine { private final WatcherOptions opts; + private final Consumer onAuthRequired; + private final CompletableFuture registered = new CompletableFuture<>(); + private final long syncId; + private boolean calledOnce; public WatcherStateMachine( @@ -48,13 +53,26 @@ public WatcherStateMachine( Connection connection, WatcherOptions opts, Timer timerService) { + this(key, syncId, callback, connection, opts, timerService, null); + } + + public WatcherStateMachine( + String key, + long syncId, + Consumer callback, + Connection connection, + WatcherOptions opts, + Timer timerService, + Consumer onAuthRequired) { this.key = key; + this.syncId = syncId; this.connection = connection; this.callback = callback; this.request = new IProtoWatch(key); this.request.setSyncId(syncId); this.opts = opts; this.timerService = timerService; + this.onAuthRequired = onAuthRequired; } @Override @@ -92,8 +110,14 @@ public boolean process(IProtoResponse message) { } else { ClientException error = new ClientException("watcher error: %s", message); registered.completeExceptionally(error); - log.warn("got error for watcher: {}", message); - opts.getErrorHandler().accept(key, error); + if (onAuthRequired != null + && message.getErrorCode() == IProtoConstant.IPROTO_ERR_AUTH_REQUIRED) { + log.debug("watcher '{}' rejected before authentication, deferred", key); + onAuthRequired.accept(this); + } else { + log.warn("got error for watcher: {}", message); + opts.getErrorHandler().accept(key, error); + } } return false; @@ -110,6 +134,10 @@ public CompletableFuture registered() { return registered; } + public long getSyncId() { + return syncId; + } + private void onSendComplete(Void r, Throwable exc) { if (exc != null) { log.warn("could not send IPROTO_WATCH packet", exc); diff --git a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java index 24f4e4b0..3fa559f3 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java +++ b/tarantool-core/src/main/java/io/tarantool/core/protocol/requests/IProtoConstant.java @@ -18,6 +18,7 @@ public interface IProtoConstant { int IPROTO_ERROR_24 = 0x31; int IPROTO_ERROR_BASE = 0x8000; int IPROTO_ERR_ACCESS_DENIED = 0x2A; + int IPROTO_ERR_AUTH_REQUIRED = 0x102; int IPROTO_ERR_CREDS_MISMATCH = 0x2F; int IPROTO_ERR_INVALID_MSGPACK = 0x20; int IPROTO_ERR_NO_SUCH_PROC = 0x21;