diff --git a/packages/postgrest/lib/src/postgrest.dart b/packages/postgrest/lib/src/postgrest.dart index a598d4399..e97e99152 100644 --- a/packages/postgrest/lib/src/postgrest.dart +++ b/packages/postgrest/lib/src/postgrest.dart @@ -1,5 +1,6 @@ import 'package:http/http.dart'; import 'package:logging/logging.dart'; +import 'package:meta/meta.dart'; import 'package:postgrest/postgrest.dart'; import 'package:postgrest/src/constants.dart'; import 'package:yet_another_json_isolate/yet_another_json_isolate.dart'; @@ -12,6 +13,8 @@ class PostgrestClient { final Client? httpClient; final YAJsonIsolate _isolate; final bool _hasCustomIsolate; + final bool retryEnabled; + final Duration Function(int attempt)? _retryDelay; final _log = Logger('supabase.postgrest'); /// To create a [PostgrestClient], you need to provide an [url] endpoint. @@ -25,16 +28,23 @@ class PostgrestClient { /// [httpClient] is optional and can be used to provide a custom http client /// /// [isolate] is optional and can be used to provide a custom isolate, which is used for heavy json computation + /// + /// [retryEnabled] controls whether automatic retries are performed for GET and + /// HEAD requests that fail with HTTP 520 or a network error. Defaults to `true`. + /// Use [PostgrestBuilder.retry] to override this per request. PostgrestClient( this.url, { Map? headers, String? schema, this.httpClient, YAJsonIsolate? isolate, + this.retryEnabled = true, + @visibleForTesting Duration Function(int attempt)? retryDelay, }) : _schema = schema, headers = {...defaultHeaders, if (headers != null) ...headers}, _isolate = isolate ?? (YAJsonIsolate()..initialize()), - _hasCustomIsolate = isolate != null { + _hasCustomIsolate = isolate != null, + _retryDelay = retryDelay { _log.config('Initialize PostgrestClient with url: $url, schema: $_schema'); _log.finest('Initialize with headers: $headers'); } @@ -65,6 +75,8 @@ class PostgrestClient { schema: _schema, httpClient: httpClient, isolate: _isolate, + clientRetryEnabled: retryEnabled, + retryDelay: _retryDelay, ); } @@ -78,6 +90,8 @@ class PostgrestClient { schema: schema, httpClient: httpClient, isolate: _isolate, + retryEnabled: retryEnabled, + retryDelay: _retryDelay, ); } @@ -108,6 +122,8 @@ class PostgrestClient { schema: _schema, httpClient: httpClient, isolate: _isolate, + clientRetryEnabled: retryEnabled, + retryDelay: _retryDelay, ).rpc(params, get); } diff --git a/packages/postgrest/lib/src/postgrest_builder.dart b/packages/postgrest/lib/src/postgrest_builder.dart index 822ac03d0..9ad7942b3 100644 --- a/packages/postgrest/lib/src/postgrest_builder.dart +++ b/packages/postgrest/lib/src/postgrest_builder.dart @@ -3,6 +3,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:core'; +import 'dart:math' as math; import 'package:http/http.dart' as http; import 'package:http/http.dart'; @@ -45,8 +46,14 @@ class PostgrestBuilder implements Future { final Client? _httpClient; final YAJsonIsolate? _isolate; final CountOption? _count; + final bool _clientRetryEnabled; + final bool? _retryEnabled; + final Duration Function(int attempt) _retryDelay; final _log = Logger('supabase.postgrest'); + static Duration _defaultRetryDelay(int attempt) => + Duration(seconds: math.min(math.pow(2, attempt).toInt(), 30)); + PostgrestBuilder({ required Uri url, required Headers headers, @@ -58,6 +65,9 @@ class PostgrestBuilder implements Future { CountOption? count, bool maybeSingle = false, PostgrestConverter? converter, + bool clientRetryEnabled = true, + bool? retryEnabled, + @visibleForTesting Duration Function(int attempt)? retryDelay, }) : _maybeSingle = maybeSingle, _method = method, _converter = converter, @@ -67,7 +77,10 @@ class PostgrestBuilder implements Future { _httpClient = httpClient, _isolate = isolate, _count = count, - _body = body; + _body = body, + _clientRetryEnabled = clientRetryEnabled, + _retryEnabled = retryEnabled, + _retryDelay = retryDelay ?? _defaultRetryDelay; PostgrestBuilder _copyWith({ Uri? url, @@ -80,6 +93,9 @@ class PostgrestBuilder implements Future { CountOption? count, bool? maybeSingle, PostgrestConverter? converter, + bool? clientRetryEnabled, + bool? retryEnabled, + Duration Function(int attempt)? retryDelay, }) { return PostgrestBuilder( url: url ?? _url, @@ -92,9 +108,21 @@ class PostgrestBuilder implements Future { count: count ?? _count, maybeSingle: maybeSingle ?? _maybeSingle, converter: converter ?? _converter, + clientRetryEnabled: clientRetryEnabled ?? _clientRetryEnabled, + retryEnabled: retryEnabled ?? _retryEnabled, + retryDelay: retryDelay ?? _retryDelay, ); } + /// Overrides the retry behavior for this specific request. + /// + /// When [enabled] is `false`, retries are disabled for this request even if + /// [PostgrestClient] was configured with `retryEnabled: true`. + /// When [enabled] is `true`, retries are enabled for this request even if + /// [PostgrestClient] was configured with `retryEnabled: false`. + PostgrestBuilder retry({required bool enabled}) => + _copyWith(retryEnabled: enabled); + PostgrestBuilder setHeader(String key, String value) { return _copyWith( headers: {..._headers, key: value}, @@ -103,13 +131,17 @@ class PostgrestBuilder implements Future { Future _execute() async { final String? method = _method; + // Work with a local copy so repeated awaits and shared-map siblings are + // not affected by per-execution header mutations (Prefer, schema headers, + // X-Retry-Count, etc.). + final execHeaders = {..._headers}; if (_count != null) { - if (_headers['Prefer'] != null) { - final oldPreferHeader = _headers['Prefer']; - _headers['Prefer'] = '$oldPreferHeader,count=${_count!.name}'; + if (execHeaders['Prefer'] != null) { + final oldPreferHeader = execHeaders['Prefer']; + execHeaders['Prefer'] = '$oldPreferHeader,count=${_count!.name}'; } else { - _headers['Prefer'] = 'count=${_count!.name}'; + execHeaders['Prefer'] = 'count=${_count!.name}'; } } @@ -121,62 +153,95 @@ class PostgrestBuilder implements Future { } final uppercaseMethod = method.toUpperCase(); - late http.Response response; if (_schema == null) { // skip } else if ([METHOD_GET, METHOD_HEAD].contains(method)) { - _headers['Accept-Profile'] = _schema!; + execHeaders['Accept-Profile'] = _schema!; } else { - _headers['Content-Profile'] = _schema!; + execHeaders['Content-Profile'] = _schema!; } if (method != METHOD_GET && method != METHOD_HEAD) { - _headers['Content-Type'] = 'application/json'; + execHeaders['Content-Type'] = 'application/json'; } final bodyStr = jsonEncode(_body); _log.finest("Request: $uppercaseMethod $_url"); + final Future Function() send; if (uppercaseMethod == METHOD_GET) { - response = await (_httpClient?.get ?? http.get)( - _url, - headers: _headers, - ); + send = () => (_httpClient?.get ?? http.get)(_url, headers: execHeaders); } else if (uppercaseMethod == METHOD_POST) { - response = await (_httpClient?.post ?? http.post)( - _url, - headers: _headers, - body: bodyStr, - ); + send = () => (_httpClient?.post ?? http.post)( + _url, + headers: execHeaders, + body: bodyStr, + ); } else if (uppercaseMethod == METHOD_PUT) { - response = await (_httpClient?.put ?? http.put)( - _url, - headers: _headers, - body: bodyStr, - ); + send = () => (_httpClient?.put ?? http.put)( + _url, + headers: execHeaders, + body: bodyStr, + ); } else if (uppercaseMethod == METHOD_PATCH) { - response = await (_httpClient?.patch ?? http.patch)( - _url, - headers: _headers, - body: bodyStr, - ); + send = () => (_httpClient?.patch ?? http.patch)( + _url, + headers: execHeaders, + body: bodyStr, + ); } else if (uppercaseMethod == METHOD_DELETE) { - response = await (_httpClient?.delete ?? http.delete)( - _url, - headers: _headers, - ); + send = () => + (_httpClient?.delete ?? http.delete)(_url, headers: execHeaders); } else if (uppercaseMethod == METHOD_HEAD) { - response = await (_httpClient?.head ?? http.head)( - _url, - headers: _headers, - ); + send = + () => (_httpClient?.head ?? http.head)(_url, headers: execHeaders); + } else { + throw StateError('Unknown HTTP method: $uppercaseMethod'); } + final response = + await _executeWithRetry(send, uppercaseMethod, execHeaders); return _parseResponse(response, method); } catch (error) { rethrow; } } + Future _executeWithRetry( + Future Function() send, + String method, + Map execHeaders, + ) async { + const maxRetries = 3; + const retryableStatusCodes = {520}; + + final effectiveRetryEnabled = _retryEnabled ?? _clientRetryEnabled; + final isRetryableMethod = method == METHOD_GET || method == METHOD_HEAD; + + if (!effectiveRetryEnabled || !isRetryableMethod) { + return send(); + } + + for (var attempt = 0; attempt <= maxRetries; attempt++) { + if (attempt > 0) { + execHeaders['X-Retry-Count'] = attempt.toString(); + } + + try { + final response = await send(); + if (!retryableStatusCodes.contains(response.statusCode) || + attempt == maxRetries) { + return response; + } + } on Exception { + if (attempt == maxRetries) rethrow; + } + + await Future.delayed(_retryDelay(attempt)); + } + + throw StateError('unreachable'); + } + /// Parse request response to json object if possible Future _parseResponse(http.Response response, String method) async { if (response.statusCode >= 200 && response.statusCode <= 299) { diff --git a/packages/postgrest/lib/src/postgrest_query_builder.dart b/packages/postgrest/lib/src/postgrest_query_builder.dart index e19439572..b927e9cf4 100644 --- a/packages/postgrest/lib/src/postgrest_query_builder.dart +++ b/packages/postgrest/lib/src/postgrest_query_builder.dart @@ -19,6 +19,8 @@ class PostgrestQueryBuilder extends RawPostgrestBuilder { String? schema, Client? httpClient, YAJsonIsolate? isolate, + bool clientRetryEnabled = true, + Duration Function(int attempt)? retryDelay, }) : super( PostgrestBuilder( url: url, @@ -27,6 +29,8 @@ class PostgrestQueryBuilder extends RawPostgrestBuilder { schema: schema, httpClient: httpClient, isolate: isolate, + clientRetryEnabled: clientRetryEnabled, + retryDelay: retryDelay, ), ); @@ -268,6 +272,8 @@ class PostgrestQueryBuilder extends RawPostgrestBuilder { method: _method, schema: _schema, isolate: _isolate, + clientRetryEnabled: _clientRetryEnabled, + retryDelay: _retryDelay, ); } } diff --git a/packages/postgrest/lib/src/postgrest_rpc_builder.dart b/packages/postgrest/lib/src/postgrest_rpc_builder.dart index 9f5190e5c..c09b16505 100644 --- a/packages/postgrest/lib/src/postgrest_rpc_builder.dart +++ b/packages/postgrest/lib/src/postgrest_rpc_builder.dart @@ -7,6 +7,8 @@ class PostgrestRpcBuilder extends RawPostgrestBuilder { String? schema, Client? httpClient, required YAJsonIsolate isolate, + bool clientRetryEnabled = true, + Duration Function(int attempt)? retryDelay, }) : super( PostgrestBuilder( url: Uri.parse(url), @@ -14,6 +16,8 @@ class PostgrestRpcBuilder extends RawPostgrestBuilder { schema: schema, httpClient: httpClient, isolate: isolate, + clientRetryEnabled: clientRetryEnabled, + retryDelay: retryDelay, ), ); diff --git a/packages/postgrest/lib/src/raw_postgrest_builder.dart b/packages/postgrest/lib/src/raw_postgrest_builder.dart index 5cf388049..059d1b152 100644 --- a/packages/postgrest/lib/src/raw_postgrest_builder.dart +++ b/packages/postgrest/lib/src/raw_postgrest_builder.dart @@ -14,6 +14,9 @@ class RawPostgrestBuilder extends PostgrestBuilder { isolate: builder._isolate, maybeSingle: builder._maybeSingle, converter: builder._converter, + clientRetryEnabled: builder._clientRetryEnabled, + retryEnabled: builder._retryEnabled, + retryDelay: builder._retryDelay, ); /// Very similar to [_copyWith], but allows changing the generics, therefore [_converter] is omitted @@ -38,6 +41,9 @@ class RawPostgrestBuilder extends PostgrestBuilder { isolate: isolate ?? _isolate, count: count ?? _count, maybeSingle: maybeSingle ?? _maybeSingle, + clientRetryEnabled: _clientRetryEnabled, + retryEnabled: _retryEnabled, + retryDelay: _retryDelay, )); } @@ -71,6 +77,9 @@ class RawPostgrestBuilder extends PostgrestBuilder { count: _count, maybeSingle: _maybeSingle, converter: converter, + clientRetryEnabled: _clientRetryEnabled, + retryEnabled: _retryEnabled, + retryDelay: _retryDelay, ); } } diff --git a/packages/postgrest/lib/src/response_postgrest_builder.dart b/packages/postgrest/lib/src/response_postgrest_builder.dart index 88efac962..16fb2ccc9 100644 --- a/packages/postgrest/lib/src/response_postgrest_builder.dart +++ b/packages/postgrest/lib/src/response_postgrest_builder.dart @@ -14,6 +14,9 @@ class ResponsePostgrestBuilder extends PostgrestBuilder { isolate: builder._isolate, maybeSingle: builder._maybeSingle, converter: builder._converter, + clientRetryEnabled: builder._clientRetryEnabled, + retryEnabled: builder._retryEnabled, + retryDelay: builder._retryDelay, ); @override @@ -49,6 +52,9 @@ class ResponsePostgrestBuilder extends PostgrestBuilder { count: _count, maybeSingle: _maybeSingle, converter: converter, + clientRetryEnabled: _clientRetryEnabled, + retryEnabled: _retryEnabled, + retryDelay: _retryDelay, ); } } diff --git a/packages/postgrest/test/retry_test.dart b/packages/postgrest/test/retry_test.dart new file mode 100644 index 000000000..0226ff82e --- /dev/null +++ b/packages/postgrest/test/retry_test.dart @@ -0,0 +1,205 @@ +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:http/http.dart'; +import 'package:postgrest/postgrest.dart'; +import 'package:test/test.dart'; + +typedef _ResponseFactory = Future Function(BaseRequest); + +_ResponseFactory _ok() => (req) async => StreamedResponse( + Stream.value(Uint8List.fromList('[]'.codeUnits)), + 200, + request: req, + headers: {'content-type': 'application/json'}, + ); + +_ResponseFactory _status(int code) => (req) async => StreamedResponse( + Stream.value( + Uint8List.fromList('{"message":"err","code":"$code"}'.codeUnits)), + code, + request: req, + headers: {'content-type': 'application/json'}, + ); + +_ResponseFactory _networkError() => + (_) async => throw const SocketException('Connection refused'); + +class _MockRetryClient extends BaseClient { + final List<_ResponseFactory> _responses; + final List requests = []; + + _MockRetryClient(this._responses); + + int get callCount => requests.length; + + @override + Future send(BaseRequest request) async { + final index = requests.length; + requests.add(request); + if (index >= _responses.length) { + throw StateError( + 'Unexpected call #${index + 1}, only ${_responses.length} configured'); + } + return _responses[index](request); + } +} + +PostgrestClient _buildClient( + _MockRetryClient mock, { + bool retryEnabled = true, +}) { + return PostgrestClient( + 'http://localhost:3000', + httpClient: mock, + retryEnabled: retryEnabled, + retryDelay: (_) => Duration.zero, + ); +} + +void main() { + group('retry logic', () { + test('GET retries on 520 then succeeds, X-Retry-Count increments', + () async { + final mock = _MockRetryClient([_status(520), _status(520), _ok()]); + final client = _buildClient(mock); + + final result = await client.from('users').select(); + + expect(result, isEmpty); + expect(mock.callCount, 3); + // Initial attempt: no header + expect(mock.requests[0].headers['x-retry-count'], isNull); + // First retry: X-Retry-Count: 1 + expect(mock.requests[1].headers['x-retry-count'], '1'); + // Second retry: X-Retry-Count: 2 + expect(mock.requests[2].headers['x-retry-count'], '2'); + }); + + test('HEAD retries on 520 then succeeds', () async { + final mock = _MockRetryClient([ + _status(520), + (req) async => StreamedResponse( + Stream.empty(), + 200, + request: req, + headers: {'content-range': '*/4'}, + ), + ]); + final client = _buildClient(mock); + + final count = await client.from('users').count(); + + expect(count, 4); + expect(mock.callCount, 2); + expect(mock.requests[1].headers['x-retry-count'], '1'); + }); + + test('POST does not retry on 520', () async { + final mock = _MockRetryClient([_status(520)]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').insert({'name': 'foo'}), + throwsA(isA()), + ); + expect(mock.callCount, 1); + }); + + test('GET does not retry on non-520 error (e.g., 400)', () async { + final mock = _MockRetryClient([_status(400)]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').select(), + throwsA(isA()), + ); + expect(mock.callCount, 1); + }); + + test('GET retries on network error (SocketException)', () async { + final mock = _MockRetryClient([_networkError(), _ok()]); + final client = _buildClient(mock); + + final result = await client.from('users').select(); + + expect(result, isEmpty); + expect(mock.callCount, 2); + expect(mock.requests[1].headers['x-retry-count'], '1'); + }); + + test('POST does not retry on network error', () async { + final mock = _MockRetryClient([_networkError()]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').insert({'name': 'foo'}), + throwsA(isA()), + ); + expect(mock.callCount, 1); + }); + + test('exhausts all 3 retries (4 total calls) then throws on 520', () async { + final mock = _MockRetryClient( + [_status(520), _status(520), _status(520), _status(520)]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').select(), + throwsA(isA()), + ); + expect(mock.callCount, 4); + }); + + test('.retry(enabled: false) disables retry per-request', () async { + final mock = _MockRetryClient([_status(520)]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').select().retry(enabled: false), + throwsA(isA()), + ); + expect(mock.callCount, 1); + }); + + test('PostgrestClient(retryEnabled: false) disables retry globally', + () async { + final mock = _MockRetryClient([_status(520)]); + final client = _buildClient(mock, retryEnabled: false); + + await expectLater( + client.from('users').select(), + throwsA(isA()), + ); + expect(mock.callCount, 1); + }); + + test('.retry(enabled: true) re-enables retry when client-level is false', + () async { + final mock = _MockRetryClient([_status(520), _ok()]); + final client = _buildClient(mock, retryEnabled: false); + + final result = await client.from('users').select().retry(enabled: true); + + expect(result, isEmpty); + expect(mock.callCount, 2); + }); + + test('GET exhausts retries on repeated network errors then rethrows', + () async { + final mock = _MockRetryClient([ + _networkError(), + _networkError(), + _networkError(), + _networkError(), + ]); + final client = _buildClient(mock); + + await expectLater( + client.from('users').select(), + throwsA(isA()), + ); + expect(mock.callCount, 4); + }); + }); +}