npgsql · roji · Sep 25, 2022 · Sep 18, 2022 · Sep 25, 2022 · Sep 25, 2022
@@ -1358,7 +1358,8 @@ internal ValueTask<IBackendMessage> ReadMessage(bool async, DataRowLoadingMode d
                             // an RFQ. Instead, the server closes the connection immediately
                             throw error;
                         }
-                        else if (PostgresErrorCodes.IsCriticalFailure(error, clusterError: false))
+
+                        if (PostgresErrorCodes.IsCriticalFailure(error, clusterError: false))
                         {
                             // Consider the connection dead
                             throw connector.Break(error);

@@ -54,6 +54,32 @@ protected override DbTransaction? DbTransaction
         set => Transaction = (NpgsqlTransaction?)value;
     }
 
+    /// <summary>
+    /// Controls whether to place error barriers between all batch commands within this batch. Default to <see langword="false" />.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    ///     By default, any exception in a command causes later commands in the batch to be skipped, and earlier commands to be rolled back.
+    ///     Enabling error barriers ensures that errors do not affect other commands in the batch.
+    /// </para>
+    /// <para>
+    ///     Note that if the batch is executed within an explicit transaction, the first error places the transaction in a failed state,
+    ///     causing all later commands to fail in any case. As a result, this option is useful mainly when there is no explicit transaction.
+    /// </para>
+    /// <para>
+    ///     At the PostgreSQL wire protocol level, this corresponds to inserting a Sync message between each command, rather than grouping
+    ///     all the batch's commands behind a single terminating Sync.
+    /// </para>
+    /// <para>
+    ///     To control error barriers on a command-by-command basis, see <see cref="NpgsqlBatchCommand.AppendErrorBarrier" />.
+    /// </para>
+    /// </remarks>
+    public bool EnableErrorBarriers
+    {
+        get => Command.EnableErrorBarriers;
+        set => Command.EnableErrorBarriers = value;
+    }
+
     /// <summary>
     /// Marks all of the batch's result columns as either known or unknown.
     /// Unknown results column are requested them from PostgreSQL in text format, and Npgsql makes no

@@ -32,6 +32,31 @@ public override string CommandText
     /// <inheritdoc cref="DbBatchCommand.Parameters"/>
     public new NpgsqlParameterCollection Parameters { get; } = new();
 
+    /// <summary>
+    /// Appends an error barrier after this batch command. Defaults to the value of <see cref="NpgsqlBatch.EnableErrorBarriers" /> on the
+    /// batch.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    ///     By default, any exception in a command causes later commands in the batch to be skipped, and earlier commands to be rolled back.
+    ///     Appending an error barrier ensures that errors from this command (or previous ones) won't cause later commands to be skipped,
+    ///     and that errors from later commands won't cause this command (or previous ones) to be rolled back).
+    /// </para>
+    /// <para>
+    ///     Note that if the batch is executed within an explicit transaction, the first error places the transaction in a failed state,
+    ///     causing all later commands to fail in any case. As a result, this option is useful mainly when there is no explicit transaction.
+    /// </para>
+    /// <para>
+    ///     At the PostgreSQL wire protocol level, this corresponds to inserting a Sync message after this command, rather than grouping
+    ///     all the batch's commands behind a single terminating Sync.
+    /// </para>
+    /// <para>
+    ///     Controlling error barriers on a command-by-command basis is an advanced feature, consider enabling error barriers for the entire
+    ///     batch via <see cref="NpgsqlBatch.EnableErrorBarriers" />.
+    /// </para>
+    /// </remarks>
+    public bool? AppendErrorBarrier { get; set; }
+
     /// <summary>
     /// The number of rows affected or retrieved.
     /// </summary>

@@ -76,6 +76,8 @@ public class NpgsqlCommand : DbCommand, ICloneable, IComponent
     internal static readonly bool EnableSqlRewriting;
 #endif
 
+    internal bool EnableErrorBarriers { get; set; }
+
     static readonly List<NpgsqlParameter> EmptyParameters = new();
 
     static readonly SingleThreadSynchronizationContext SingleThreadSynchronizationContext = new("NpgsqlRemainingAsyncSendWorker");
@@ -961,12 +963,14 @@ internal Task Write(NpgsqlConnector connector, bool async, bool flush, Cancellat
 
         async Task WriteExecute(NpgsqlConnector connector, bool async, bool flush, CancellationToken cancellationToken)
         {
+            NpgsqlBatchCommand? batchCommand = null;
+
             for (var i = 0; i < InternalBatchCommands.Count; i++)
             {
                 // The following is only for deadlock avoidance when doing sync I/O (so never in multiplexing)
                 ForceAsyncIfNecessary(ref async, i);
 
-                var batchCommand = InternalBatchCommands[i];
+                batchCommand = InternalBatchCommands[i];
                 var pStatement = batchCommand.PreparedStatement;
 
                 Debug.Assert(batchCommand.FinalCommandText is not null);
@@ -1000,11 +1004,17 @@ await connector.WriteBind(
 
                 await connector.WriteExecute(0, async, cancellationToken);
 
+                if (batchCommand.AppendErrorBarrier ?? EnableErrorBarriers)
+                    await connector.WriteSync(async, cancellationToken);
+
                 if (pStatement != null)
                     pStatement.LastUsed = DateTime.UtcNow;
             }
 
-            await connector.WriteSync(async, cancellationToken);
+            if (batchCommand is null || !(batchCommand.AppendErrorBarrier ?? EnableErrorBarriers))
+            {
+                await connector.WriteSync(async, cancellationToken);
+            }
 
             if (flush)
                 await connector.Flush(async, cancellationToken);

@@ -9,6 +9,7 @@
 using System.IO;
 using System.Linq;
 using System.Runtime.CompilerServices;
+using System.Runtime.ExceptionServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -283,9 +284,24 @@ async Task<bool> Read(bool async, CancellationToken cancellationToken = default)
                 throw new ArgumentOutOfRangeException();
             }
 
-            var msg2 = await ReadMessage(async);
-            ProcessMessage(msg2);
-            return msg2.Code == BackendMessageCode.DataRow;
+            var msg = await ReadMessage(async);
+
+            switch (msg.Code)
+            {
+            case BackendMessageCode.DataRow:
+                ProcessMessage(msg);
+                return true;
+
+            case BackendMessageCode.CommandComplete:
+            case BackendMessageCode.EmptyQueryResponse:
+                ProcessMessage(msg);
+                if (_statements[StatementIndex].AppendErrorBarrier ?? Command.EnableErrorBarriers)
+                    Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
+                return false;
+
+            default:
+                throw Connector.UnexpectedMessageReceived(msg.Code);
+            }
         }
         catch
         {
@@ -335,10 +351,11 @@ public override bool NextResult() => (_isSchemaOnly ? NextResultSchemaOnly(false
     /// <returns>A task representing the asynchronous operation.</returns>
     public override Task<bool> NextResultAsync(CancellationToken cancellationToken)
     {
-        using (NoSynchronizationContextScope.Enter())
-            return _isSchemaOnly
-                ? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken)
-                : NextResult(async: true, cancellationToken: cancellationToken);
+        using var _ = NoSynchronizationContextScope.Enter();
+
+        return _isSchemaOnly
+            ? NextResultSchemaOnly(async: true, cancellationToken: cancellationToken)
+            : NextResult(async: true, cancellationToken: cancellationToken);
     }
 
     /// <summary>
@@ -370,7 +387,12 @@ async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationTo
                     case BackendMessageCode.CommandComplete:
                     case BackendMessageCode.EmptyQueryResponse:
                         ProcessMessage(completedMsg);
+
+                        if (_statements[StatementIndex].AppendErrorBarrier ?? Command.EnableErrorBarriers)
+                            Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
+
                         break;
+
                     default:
                         continue;
                     }
@@ -472,6 +494,10 @@ async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationTo
                     }
 
                     ProcessMessage(msg);
+
+                    if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers)
+                        Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
+
                     continue;
                 }
 
@@ -494,30 +520,32 @@ async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationTo
                 switch (msg.Code)
                 {
                 case BackendMessageCode.DataRow:
+                    return true;
                 case BackendMessageCode.CommandComplete:
-                    break;
+                    if (statement.AppendErrorBarrier ?? Command.EnableErrorBarriers)
+                        Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
+                    return true;
                 default:
                     throw Connector.UnexpectedMessageReceived(msg.Code);
                 }
-
-                return true;
             }
 
             // There are no more queries, we're done. Read the RFQ.
-            ProcessMessage(Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector));
+            if (_statements.Count == 0 || !(_statements[_statements.Count - 1].AppendErrorBarrier ?? Command.EnableErrorBarriers))
+                Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
+
+            State = ReaderState.Consumed;
             RowDescription = null;
             return false;
         }
         catch (Exception e)
         {
-            State = ReaderState.Consumed;
-
             // Reference the triggering statement from the exception
             if (e is PostgresException postgresException && StatementIndex >= 0 && StatementIndex < _statements.Count)
             {
                 postgresException.BatchCommand = _statements[StatementIndex];
 
-                // Prevent the command or batch from by recycled (by the connection) when it's disposed. This is important since
+                // Prevent the command or batch from being recycled (by the connection) when it's disposed. This is important since
                 // the exception is very likely to escape the using statement of the command, and by that time some other user may
                 // already be using the recycled instance.
                 if (!Command.IsWrappedByBatch)
@@ -526,9 +554,8 @@ async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationTo
                 }
             }
 
-            // An error means all subsequent statements were skipped by PostgreSQL.
-            // If any of them were being prepared, we need to update our bookkeeping to put
-            // them back in unprepared state.
+            // For the statement that errored, if it was being prepared we need to update our bookkeeping to put them back in unprepared
+            // state.
             for (; StatementIndex < _statements.Count; StatementIndex++)
             {
                 var statement = _statements[StatementIndex];
@@ -537,8 +564,33 @@ async Task<bool> NextResult(bool async, bool isConsuming = false, CancellationTo
                     statement.IsPreparing = false;
                     statement.PreparedStatement!.AbortPrepare();
                 }
+
+                // In normal, non-isolated batching, we've consumed the result set and are done.
+                // However, if the command has error barrier, we now have to consume results from the commands after it (unless it's the
+                // last one).
+                // Note that Consume calls NextResult (this method) recursively, the isConsuming flag tells us we're in this mode.
+                if ((statement.AppendErrorBarrier ?? Command.EnableErrorBarriers) && StatementIndex < _statements.Count - 1)
+                {
+                    if (isConsuming)
+                        throw;
+                    switch (State)
+                    {
+                    case ReaderState.Consumed:
+                    case ReaderState.Closed:
+                    case ReaderState.Disposed:
+                        // The exception may have caused the connector to break (e.g. I/O), and so the reader is already closed.
+                        break;
+                    default:
+                        // We provide Consume with the first exception which we've just caught.
+                        // If it encounters other exceptions while consuming the rest of the result set, it will raise an AggregateException,
+                        // otherwise it will rethrow this first exception.
+                        await Consume(async, firstException: e);
+                        break; // Never reached, Consume always throws above
+                    }
+                }
             }
 
+            State = ReaderState.Consumed;
             throw;
         }
     }
@@ -672,8 +724,9 @@ async Task<bool> NextResultSchemaOnly(bool async, bool isConsuming = false, Canc
             // There are no more queries, we're done. Read to the RFQ.
             if (!_statements.All(s => s.IsPrepared))
             {
-                ProcessMessage(Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector));
+                Expect<ReadyForQueryMessage>(await Connector.ReadMessage(async), Connector);
                 RowDescription = null;
+                State = ReaderState.Consumed;
             }
 
             return false;
@@ -748,10 +801,6 @@ internal void ProcessMessage(IBackendMessage msg)
             State = ReaderState.BetweenResults;
             return;
 
-        case BackendMessageCode.ReadyForQuery:
-            State = ReaderState.Consumed;
-            return;
-
         default:
             throw new Exception("Received unexpected backend message of type " + msg.Code);
         }
@@ -901,14 +950,44 @@ public override int FieldCount
     /// Consumes all result sets for this reader, leaving the connector ready for sending and processing further
     /// queries
     /// </summary>
-    async Task Consume(bool async)
+    async Task Consume(bool async, Exception? firstException = null)
     {
-        // Skip over the other result sets. Note that this does tally records affected
-        // from CommandComplete messages, and properly sets state for auto-prepared statements
-        if (_isSchemaOnly)
-            while (await NextResultSchemaOnly(async, isConsuming: true)) {}
-        else
-            while (await NextResult(async, isConsuming: true)) {}
+        var exceptions = firstException is null ? null : new List<Exception> { firstException };
+
+        // Skip over the other result sets. Note that this does tally records affected from CommandComplete messages, and properly sets
+        // state for auto-prepared statements
+        while (true)
+        {
+            try
+            {
+                if (!(_isSchemaOnly
+                        ? await NextResultSchemaOnly(async, isConsuming: true)
+                        : await NextResult(async, isConsuming: true)))
+                {
+                    break;
+                }
+            }
+            catch (Exception e)
+            {
+                exceptions ??= new();
+                exceptions.Add(e);
+            }
+        }
+
+        Debug.Assert(exceptions?.Count != 0);
+
+        switch (exceptions?.Count)
+        {
+        case null:
+            return;
+        case 1:
+            ExceptionDispatchInfo.Capture(exceptions[0]).Throw();
+            return;
+        default:
+            throw new NpgsqlException(
+                "Multiple exceptions occurred when consuming the result set",
+                new AggregateException(exceptions));
+        }
     }
 
     /// <summary>

@@ -20,7 +20,7 @@ class PostgresMinimalDatabaseInfoFactory : INpgsqlDatabaseInfoFactory
 
 class PostgresMinimalDatabaseInfo : PostgresDatabaseInfo
 {
-    static PostgresType[]? TypesWithMultiranges, TypesWithoutMultiranges;
+    static PostgresType[]? _typesWithMultiranges, _typesWithoutMultiranges;
 
     static PostgresType[] CreateTypes(bool withMultiranges)
         => typeof(NpgsqlDbType).GetFields()
@@ -50,8 +50,8 @@ static PostgresType[] CreateTypes(bool withMultiranges)
 
     protected override IEnumerable<PostgresType> GetTypes()
         => SupportsMultirangeTypes
-            ? TypesWithMultiranges ??= CreateTypes(withMultiranges: true)
-            : TypesWithoutMultiranges ??= CreateTypes(withMultiranges: false);
+            ? _typesWithMultiranges ??= CreateTypes(withMultiranges: true)
+            : _typesWithoutMultiranges ??= CreateTypes(withMultiranges: false);
 
     internal PostgresMinimalDatabaseInfo(NpgsqlConnector conn)
         : base(conn)

@@ -1,4 +1,8 @@
 #nullable enable
+Npgsql.NpgsqlBatch.EnableErrorBarriers.get -> bool
+Npgsql.NpgsqlBatch.EnableErrorBarriers.set -> void
+Npgsql.NpgsqlBatchCommand.AppendErrorBarrier.get -> bool?
+Npgsql.NpgsqlBatchCommand.AppendErrorBarrier.set -> void
 Npgsql.NpgsqlLoggingConfiguration
 static Npgsql.NpgsqlLoggingConfiguration.InitializeLogging(Microsoft.Extensions.Logging.ILoggerFactory! loggerFactory, bool parameterLoggingEnabled = false) -> void
 *REMOVED*Npgsql.NpgsqlConnection.Settings.get -> Npgsql.NpgsqlConnectionStringBuilder!