fix(tarantool-core): send IPROTO_WATCH after authentication

.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

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

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<Integer> serverProtocolVersion;
   private CompletableFuture<EnumSet<IProtoFeature>> serverFeatures;
+  private volatile boolean authorized;
   private Set<IProtoFeature> clientFeaturesEnum;
   private List<Integer> clientFeaturesList;
   private LongTaskTimer requestTimer;
@@ -185,11 +186,10 @@ public CompletableFuture<Void> connect(InetSocketAddress address, long timeoutMs
   public CompletableFuture<Void> 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<IProtoResponse> 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<IProtoResponse> execute(
 
   @Override
   public CompletableFuture<IProtoResponse> 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));

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<WatcherStateMachine> onAuthRequired;
+
   private final CompletableFuture<Void> 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<IProtoResponse> callback,
+      Connection connection,
+      WatcherOptions opts,
+      Timer timerService,
+      Consumer<WatcherStateMachine> 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<Void> 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);

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;