add support in Pool retry to run again the query if the database is restarted

[insinfo]

It would be very useful to have a retry if the query throws an error if the database is restarted

[isoos]

Yes, I'm planning to do this, however, I'm not sure what the best API choice would be:

• putting the extra parameter(s) in Pool.run[Tx]() https://pub.dev/documentation/postgres/latest/postgres/Pool/run.html
• putting the extra parameter(s) in Session.execute (and prepare+run)
• putting the extra parameter(s) into one of the *Settings object

Any insight on how this feels better? /cc @simolus3

[insinfo]

I think the ideal would be to have this option in PoolSettings

it seems that in Java BaseObjectPoolConfig has a retry option

"In this particular case the PostgreSQL connection is telling you that the server was shut down after the connection was created. DBCP connection pool does not handle this condition with it's default configuration.
Even if you set validationQuery parameter to something like SELECT 1, it will not be used, unless you also set at least one of the testOnXXXX parameters.
I usually set both testOnCreate and testOnBorrow to true."

https://docs.progress.com/pt-BR/bundle/datadirect-connect-jdbc-51/page/Specifying-Connection-Retry_7.html

It seems that in C# SQL Server also has a retry option

https://learn.microsoft.com/en-us/sql/connect/ado-net/configurable-retry-logic-sqlclient-introduction?view=sql-server-ver16
https://commons.apache.org/proper/commons-pool/apidocs/org/apache/commons/pool2/impl/BaseObjectPoolConfig.html

At the moment I'm doing something like this, a workaround

import 'package:eloquent/eloquent.dart';
import 'package:rava_backend/rava_backend.dart';
import 'package:shelf/shelf.dart';
import 'package:postgres/postgres.dart' as postgres;


final auditoriaService = AuditoriaService();

class AuditoriaService {
  late Connection conn;
  late AuditoriaRepository repo;
  int _retryCount = 0;
  final int _retryLimit = 10;

  AuditoriaService();

  Future<void> init() async {
    conn = await DBLayer().connect('auditoria');
    repo = AuditoriaRepository(conn);
    _retryCount = 0;
  }

  Future<void> acaoInserir(String objeto, Request req) async {
    await _insert(Auditoria.acaoInserir(
        usuario: req.token?.loginName ?? '', objeto: objeto));
  }

  Future<void> acaoAtualizar(String objeto, Request req) async {
    await _insert(Auditoria.acaoAtualizar(
        usuario: req.token?.loginName ?? '', objeto: objeto));
  }

  Future<void> acaoRemover(String objeto, Request req) async {
    await _insert(Auditoria.acaoRemover(
        usuario: req.token?.loginName ?? '', objeto: objeto));
  }


  Future<void> _insert(Auditoria auditoria) async {
    try {
      await repo.insert(auditoria);
    } catch (e) {
      
      if (e is postgres.PgException &&
          e.toString().contains('connection is not open') &&
          _retryCount < _retryLimit) {
        await init();
        _retryCount++;
        print('restart');
        await repo.insert(auditoria);
      } else {
        rethrow;
      }
  
    }
  }
}

[simolus3]

Any insight on how this feels better? /cc @simolus3

Given that this is likely pool-specific, putting it on Session.execute feels a bit weird. I also think this should be part of PoolSettings (perhaps with an optional parameter on Pool.run to override the default behavior in a limited scope, but I'm not sure how useful that would be).

[isoos]

I've reviewed my prior use of package:postgres_pool, and it turns out I've used these a lot more than I've remembered for optimistic transactions (this is a codebase I don't plan to touch or migrate in the near future):

RetryOptions? retryOptions,
  FutureOr<R> Function()? orElse,
  FutureOr<bool> Function(Exception)? retryIf,

https://pub.dev/documentation/postgres_pool/latest/postgres_pool/PgPool/runTx.html

It is usually specific to the kind of processing I want to do, and more strangely, it specifically did not retry on network errors - which now looks strange, but the processes are restarted via an external supervisord.

I think the orElse should go into the method parameters, possibly the retryIf too, although that could also go into ...Settings... however, run/runTx uses SessionSettings and TransactionSettings, we may need to introduce PoolSessionSettings and/or PoolTransactionSettings for them if this should be included there.

Also to consider: we have locality parameter outside of the settings, which is also very specific to the processing we want to run inside the run(tx) callback.

[isoos]

Iterating a bit more on this: we may split the connection-related retries (which in the pool context may just land on a different instance) and the application-logic retries (which may be run on the same instance).

1. I think the connection-related retry settings may go to into the PoolSettings object, as it will be applied over the entire lifecycle of the pool usage. I'm unsure if we may allow an override it in a potential PoolSessionSettings or PoolTransactionSettings object, but that is certainly a future possibility there.

2. I think the application-logic retries should not go to the PoolSettings object, or even if they go, we must have a run[Tx]-level override for them.

3. If these are implemented separately, we may retry an operation N * M times (assuming N connection retry and M application-level retry). This may not be bad, just a side-effect that we need to document and be explicit about. Maybe it is easier to merge the two options and have only max(N, M) retries though, over a single loop. I'm going back and forth on what is better, feel free to weight any opinion on that.

4. I'm also a bit on the side to not expose package:retry APIs directly on the package:postgres API, only to rely on it in the implementation. Exposing fewer options helps to keep the complexity somewhat lower.

[busslina]

Hi, is there any example of a Pool usage available?

[isoos]

Hi, is there any example of a Pool usage available?

What are you looking for? It is mostly a drop-in for connections and their renewals, so there is not much to it in terms of how to use them.

Maybe the tests can give you further hints?
https://github.com/isoos/postgresql-dart/blob/master/test/pool_test.dart

[busslina]

I reformulated the question here

[ryanheise]

Just to mention, there is an "unreferenced" Retry class in the API (lib/postgres.dart):

/// The settings that control the retry of [SessionExecutor.run] and [SessionExecutor.runTx] methods.
class Retry<R> {
  /// The maximum number of attempts to run the operation.
  final int maxAttempts;

  final FutureOr<R> Function()? orElse;
  final FutureOr<bool> Function(Exception)? retryIf;

  Retry({
    required this.maxAttempts,
    this.orElse,
    this.retryIf,
  });
}

(I don't have any ideas to contribute on what the API could be, except that implementing exponential backoff would be nice.)

[isoos]

@ryanheise Yes, that's something I wanted to include in the run / runTx methods, although I think it was stalled because of an API indecision (whether it becomes an extra parameter or part of the *Settings object, and if the later, how does it make sense in all context (incl. pool connection retry vs. run fn retry). Any preferences or opinions there?

[hurrba]

any updates on this?
would love to help contribute, if any API decision has been made?

wouldn't it make sense to have this logic in withConnection()?

[isoos]

wouldn't it make sense to have this logic in withConnection()?

In case of the pool, run and runTx are calls to withConnection(), so there is not much difference there, and in the non-pool case they could be still meaningful in case of transaction or optimistic query retries.

[hurrba]

for anyone intereseted, I ended up creating an extension as I also wanted to handle other exceptions:

extension SessionExtension on Session {
  Future<Result> executeAndCatch(
    String sql, {
    Object? parameters,
  }) async {
    return await _execute(sql, parameters: parameters);
  }

  Future<Result> _execute(
    String sql, {
    Object? parameters,
    int retryCount = 0,
  }) async {
    try {
      return await execute(Sql.named(sql), parameters: parameters);
    } catch (e) {
      switch (e) {
        case ServerException(code: '23505', message: final message):
          throw DuplicateKeyException(message);
        case ServerException(code: '23503', message: final message):
          throw ForeignKeyViolationException(message);
        case PgException(message: 'Attempting to execute query, but connection is not open.', severity: Severity.error) ||
             SocketException(osError: OSError(message: 'Connection refused', errorCode: 111)) || 
            SocketException(osError: OSError(message : 'Connection reset by peer', errorCode: 104):
          if (retryCount < 3) {
            int backoff = 1 << retryCount;

            logger.info('Connection is not open, retrying in $backoff seconds...');

            await Future.delayed(Duration(seconds: backoff));
            return await _execute(
              sql,
              parameters: parameters,
              retryCount: retryCount + 1,
            );
          }
          rethrow;
        default:
          logger.severe('Error executing query:\n$sql', e);
          rethrow;
      }
    }
  }
}

[isoos]

I'm curious: we have had a lot of bug fixes in the recent months, where are we standing with this issue? Is there any suggestion we could adopt?

@hurrba: I like the ForeignKeyViolationException and DuplicateKeyException subtypes, would you be interested to submit them as PRs?