#879 Statement close of a leaked statement did not detect fatal connection errors

Also fixes an NPE when the connection was terminated due to a fatal connection error.

Author: mrotteveel

src/main/org/firebirdsql/gds/ng/wire/AbstractFbWireDatabase.java

@@ -6,6 +6,7 @@
 import org.firebirdsql.gds.impl.wire.XdrInputStream;
 import org.firebirdsql.gds.impl.wire.XdrOutputStream;
 import org.firebirdsql.gds.ng.*;
+import org.jspecify.annotations.NullMarked;
 
 import java.io.IOException;
 import java.sql.SQLException;
@@ -210,9 +211,52 @@ public final Response readResponse(WarningMessageCallback warningCallback) throw
         return wireOperations.readResponse(warningCallback);
     }
 
+    /**
+     * {@inheritDoc}
+     * <p>
+     * The implementation decorates {@code deferredAction} to ensure this database instance is notified of exceptions.
+     * This can result in double notifications if the caller also notifies exceptions and this database instance is
+     * registered as an exception listener for those notifications (this is not harmful).
+     * </p>
+     */
     @Override
     public final void enqueueDeferredAction(DeferredAction deferredAction) throws SQLException {
-        wireOperations.enqueueDeferredAction(deferredAction);
+        wireOperations.enqueueDeferredAction(decorateWithExceptionNotification(deferredAction));
+    }
+
+    @NullMarked
+    private DeferredAction decorateWithExceptionNotification(DeferredAction deferredAction) {
+        return deferredAction instanceof ExceptionNotifyingDeferredAction
+                // Don't decorate again
+                ? deferredAction
+                : new ExceptionNotifyingDeferredAction(deferredAction);
+    }
+
+    @NullMarked
+    private final class ExceptionNotifyingDeferredAction extends DeferredAction.DelegatingDeferredAction {
+
+        ExceptionNotifyingDeferredAction(DeferredAction delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public void onException(Exception exception) {
+            try {
+                super.onException(exception);
+            } finally {
+                exceptionListenerDispatcher.errorOccurred(toReadSQLException(exception));
+            }
+        }
+
+        private static SQLException toReadSQLException(Exception exception) {
+            if (exception instanceof SQLException sqle) {
+                return sqle;
+            } else if (exception instanceof IOException ioe) {
+                return FbExceptionBuilder.ioReadError(ioe);
+            }
+            return new SQLException(exception);
+        }
+
     }
 
     @Override

src/main/org/firebirdsql/gds/ng/wire/AbstractFbWireStatement.java

@@ -229,18 +229,6 @@ private static final class CleanupAction implements Runnable, StatementListener,
         private static final AtomicReferenceFieldUpdater<CleanupAction, FbWireDatabase> databaseUpdater =
                 AtomicReferenceFieldUpdater.newUpdater(CleanupAction.class, FbWireDatabase.class, "database");
 
-        private static final DeferredAction CLEANUP_FREE_DEFERRED_ACTION = new DeferredAction() {
-            @Override
-            public void processResponse(Response response) {
-                // nothing to do
-            }
-
-            @Override
-            public boolean requiresSync() {
-                return true;
-            }
-        };
-
         private final int handle;
         private volatile FbWireDatabase database;
 
@@ -296,7 +284,7 @@ public void run() {
                     xdrOut.flush();
                 });
                 // TODO: This may process deferred actions on the cleaner thread, we may want to change that
-                database.enqueueDeferredAction(CLEANUP_FREE_DEFERRED_ACTION);
+                database.enqueueDeferredAction(DeferredAction.NO_OP_INSTANCE);
             } catch (SQLException | IOException e) {
                 System.getLogger(getClass().getName()).log(System.Logger.Level.TRACE,
                         "Ignored exception during statement clean up", e);

src/main/org/firebirdsql/gds/ng/wire/AbstractWireOperations.java

@@ -130,7 +130,7 @@ public final Response readOperationResponse(int operationCode, WarningMessageCal
     }
 
     @Override
-    public final int readNextOperation() throws IOException {
+    public final int readNextOperation() throws SQLException, IOException {
         try (LockCloseable ignored = withLock()) {
             processDeferredActions();
             return connection.readNextOperation();

src/main/org/firebirdsql/gds/ng/wire/DeferredAction.java

@@ -1,24 +1,28 @@
-// SPDX-FileCopyrightText: Copyright 2014-2023 Mark Rotteveel
+// SPDX-FileCopyrightText: Copyright 2014-2026 Mark Rotteveel
 // SPDX-License-Identifier: LGPL-2.1-or-later OR BSD-3-Clause
 package org.firebirdsql.gds.ng.wire;
 
 import org.firebirdsql.gds.ng.DeferredResponse;
 import org.firebirdsql.gds.ng.WarningMessageCallback;
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
 
 import java.util.function.Consumer;
 import java.util.function.Function;
 
 import static java.lang.System.Logger.Level.DEBUG;
+import static java.util.Objects.requireNonNull;
 
 /**
  * Interface for processing deferred responses from the server.
  * <p>
- * This interfaces is used in protocol 11 or higher.
+ * This interface is used in protocol 11 or higher.
  * </p>
  *
  * @author Mark Rotteveel
  * @since 3.0
  */
+@NullMarked
 public interface DeferredAction {
 
     /**
@@ -62,7 +66,7 @@ default void onException(Exception exception) {
      *
      * @return warning callback to use when executing this deferred action, {@code null} signals to use the default
      */
-    default WarningMessageCallback getWarningMessageCallback() {
+    default @Nullable WarningMessageCallback getWarningMessageCallback() {
         return null;
     }
 
@@ -97,25 +101,99 @@ default boolean requiresSync() {
      * @return deferred action
      */
     static <T> DeferredAction wrapDeferredResponse(DeferredResponse<T> deferredResponse,
-            Function<Response, T> responseMapper, WarningMessageCallback warningMessageCallback,
+            Function<Response, T> responseMapper, @Nullable WarningMessageCallback warningMessageCallback,
             Consumer<Exception> exceptionConsumer, boolean requiresSync) {
-        return new DeferredAction() {
+        return builder()
+                .withProcessResponse(response -> deferredResponse.onResponse(responseMapper.apply(response)))
+                .withOnException(exception -> {
+                    try {
+                        deferredResponse.onException(exception);
+                    } finally {
+                        exceptionConsumer.accept(exception);
+                    }
+                })
+                .withWarningMessageCallback(warningMessageCallback)
+                .withRequiresSync(requiresSync)
+                .build();
+    }
+
+    /**
+     * @return builder for a deferred action instance.
+     * @since 7
+     */
+    static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * Builder for lambda-based deferred actions.
+     *
+     * @since 7
+     */
+    final class Builder {
+
+        private static final Consumer<Response> RESPONSE_NO_OP = r -> {};
+        private static final Consumer<Exception> EXCEPTION_NO_OP = e ->
+                System.getLogger(DeferredAction.class.getName()).log(DEBUG, "Exception in processDeferredActions", e);
+
+        private Consumer<Response> processResponse = RESPONSE_NO_OP;
+        private Consumer<Exception> onException = EXCEPTION_NO_OP;
+        private @Nullable WarningMessageCallback warningMessageCallback;
+        private boolean requiresSync;
+
+        private Builder() {
+        }
+
+        public Builder withProcessResponse(Consumer<Response> processResponse) {
+            this.processResponse = processResponse;
+            return this;
+        }
+
+        public Builder withOnException(Consumer<Exception> onException) {
+            this.onException = onException;
+            return this;
+        }
+
+        public Builder withWarningMessageCallback(@Nullable WarningMessageCallback warningMessageCallback) {
+            this.warningMessageCallback = warningMessageCallback;
+            return this;
+        }
+
+        public Builder withRequiresSync(boolean requiresSync) {
+            this.requiresSync = requiresSync;
+            return this;
+        }
+
+        public DeferredAction build() {
+            if (processResponse == RESPONSE_NO_OP && onException == EXCEPTION_NO_OP && warningMessageCallback == null
+                    && !requiresSync) {
+                return NO_OP_INSTANCE;
+            }
+
+            return new DeferredActionImpl(processResponse, onException, warningMessageCallback, requiresSync);
+        }
+
+        /**
+         * Lambda-based deferred action implementation.
+         *
+         * @since 7
+         */
+        private record DeferredActionImpl(Consumer<Response> processResponse, Consumer<Exception> onException,
+                @Nullable WarningMessageCallback warningMessageCallback, boolean requiresSync)
+                implements DeferredAction {
+
             @Override
             public void processResponse(Response response) {
-                deferredResponse.onResponse(responseMapper.apply(response));
+                processResponse.accept(response);
             }
 
             @Override
             public void onException(Exception exception) {
-                try {
-                    deferredResponse.onException(exception);
-                } finally {
-                    exceptionConsumer.accept(exception);
-                }
+                onException.accept(exception);
             }
 
             @Override
-            public WarningMessageCallback getWarningMessageCallback() {
+            public @Nullable WarningMessageCallback getWarningMessageCallback() {
                 return warningMessageCallback;
             }
 
@@ -124,7 +202,46 @@ public boolean requiresSync() {
                 return requiresSync;
             }
 
-        };
+        }
+
+    }
+
+    /**
+     * Deferred action implementation that delegates to another deferred action.
+     * <p>
+     * This class is intended as a base class for implementations that want to decorate method calls. To decorate it,
+     * subclass this class, and override the method, and ensure you call {@code super.<overridden-method>} in such a
+     * way that it is always called, even if the decoration fails.
+     * </p>
+     * @since 7
+     */
+    abstract class DelegatingDeferredAction implements DeferredAction {
+
+        private final DeferredAction delegate;
+
+        DelegatingDeferredAction(DeferredAction delegate) {
+            this.delegate = requireNonNull(delegate, "delegate");
+        }
+
+        @Override
+        public void processResponse(Response response) {
+            delegate.processResponse(response);
+        }
+
+        @Override
+        public void onException(Exception exception) {
+            delegate.onException(exception);
+        }
+
+        @Override
+        public @Nullable WarningMessageCallback getWarningMessageCallback() {
+            return delegate.getWarningMessageCallback();
+        }
+
+        @Override
+        public boolean requiresSync() {
+            return delegate.requiresSync();
+        }
     }
 
 }

src/main/org/firebirdsql/gds/ng/wire/FbWireOperations.java

@@ -59,11 +59,13 @@ public interface FbWireOperations {
      * </p>
      *
      * @return next operation
-     * @throws java.io.IOException
+     * @throws SQLException
+     *         when the connection is closed
+     * @throws IOException
      *         for errors reading the operation from the connection
      * @since 6
      */
-    int readNextOperation() throws IOException;
+    int readNextOperation() throws SQLException, IOException;
 
     /**
      * Reads the response from the server when the operation code has already been read.

src/main/org/firebirdsql/gds/ng/wire/WireConnection.java

@@ -635,11 +635,14 @@ private FbWireOperations getCryptKeyCallbackWireOperations() {
      * Reads the next operation code. Skips all {@link org.firebirdsql.gds.impl.wire.WireProtocolConstants#op_dummy}
      * codes received.
      *
-     * @return Operation code
+     * @return operation code
+     * @throws SQLException
+     *         when the connection is closed
      * @throws IOException
      *         if an error occurs while reading from the underlying InputStream
      */
-    public final int readNextOperation() throws IOException {
+    public final int readNextOperation() throws SQLException, IOException {
+        final XdrInputStream xdrIn = getXdrStreamAccess().getXdrIn();
         int op;
         do {
             op = xdrIn.readInt();

src/test/org/firebirdsql/management/FBMaintenanceManagerTest.java

@@ -10,6 +10,7 @@
 import org.firebirdsql.common.extension.RunEnvironmentExtension.EnvironmentRequirement;
 import org.firebirdsql.common.extension.UsesDatabaseExtension;
 import org.firebirdsql.gds.ISCConstants;
+import org.firebirdsql.gds.JaybirdErrorCodes;
 import org.firebirdsql.gds.TransactionParameterBuffer;
 import org.firebirdsql.gds.ng.FbDatabase;
 import org.firebirdsql.gds.ng.FbTransaction;
@@ -179,7 +180,7 @@ void testSetBadDialect() {
     @ValueSource(booleans = { true, false })
     void testForcedShutdown(boolean autoCommit) throws Exception {
         assumeThat("Test doesn't work correctly under embedded", GDS_TYPE, not(isEmbeddedType()));
-        try (Connection conn = getConnectionViaDriverManager()) {
+        try (var conn = getConnectionViaDriverManager()) {
             createTestTable();
 
             conn.setAutoCommit(autoCommit);
@@ -191,7 +192,8 @@ void testForcedShutdown(boolean autoCommit) throws Exception {
 
                 SQLException exception = assertThrows(SQLException.class, () -> stmt.executeQuery(sql));
                 assertThat(exception, errorCode(oneOf(
-                        ISCConstants.isc_shutdown, ISCConstants.isc_att_shutdown, ISCConstants.isc_net_read_err)));
+                        ISCConstants.isc_shutdown, ISCConstants.isc_att_shutdown, ISCConstants.isc_net_read_err,
+                        JaybirdErrorCodes.jb_connectionClosed)));
             } finally {
                 closeQuietly(stmt);
             }