perf(network): reuse HTTP connections for browsing and range reads

Pool bounded read clients by connection credentials and certificate pin, invalidate them after configuration changes, and abort timed-out requests. Verify connection reuse, concurrent range limits, error recovery and credential rotation.
This commit is contained in:
zarzet committed 2026-10-10 23:22:32 +07:00
1 parent e9b7e84113
commit 0d70e413ed
2 files changed
+207 -34

No files matched your search

+103 -34
View File
@@ -206,6 +206,17 @@ class NetworkStorageService {
Future<HttpServer>? _server;
Future<void> _mutation = Future.value();
final _routes = <String, ({String id, String path})>{};
final _readClients =
<
({
String id,
String address,
String username,
String password,
String? certificate,
}),
HttpClient
>{};
static String newId() {
final random = Random.secure();
@@ -249,6 +260,7 @@ class NetworkStorageService {
next.add(connection);
await _write(jsonEncode(next.map((c) => c.toJson()).toList()));
_connections = List.unmodifiable(next);
_closeReadClients(connection.id);
});
Future<void> remove(String id) => _mutate(() async {
@@ -256,6 +268,7 @@ class NetworkStorageService {
await _write(jsonEncode(next.map((c) => c.toJson()).toList()));
_connections = List.unmodifiable(next);
_routes.removeWhere((_, route) => route.id == id);
_closeReadClients(id);
});
Future<NetworkConnection> connection(String id) async =>
@@ -331,6 +344,35 @@ class NetworkStorageService {
..idleTimeout = timeout
..autoUncompress = false;
HttpClient _readClient(NetworkConnection connection) {
// Pool only read requests with the same credentials and certificate pin.
// Uploads retain their own scoped client and cancellation lifecycle.
final key = (
id: connection.id,
address: connection.address,
username: connection.username,
password: connection.password,
certificate: connection.trustedCertificateSha256,
);
final client =
_readClients.remove(key) ?? (_client()..maxConnectionsPerHost = 6);
while (_readClients.length >= 8) {
_readClients.remove(_readClients.keys.first)!.close();
}
_readClients[key] = client;
return client;
}
void _closeReadClients(String id) {
for (final key in _readClients.keys.where((key) => key.id == id).toList()) {
// Let in-flight reads finish; new requests use the updated configuration.
_readClients.remove(key)!.close();
}
}
Future<void> _discardResponse(HttpClientResponse response) =>
response.listen((_) {}).cancel();
Future<HttpClientResponse> _request(
HttpClient client,
NetworkConnection c,
@@ -363,7 +405,18 @@ class NetworkStorageService {
};
final HttpClientRequest request;
try {
request = await client.openUrl(method, uri).timeout(timeout);
var expired = false;
final opening = client.openUrl(method, uri).then((request) {
if (expired) request.abort();
return request;
});
request = await opening.timeout(
timeout,
onTimeout: () {
expired = true;
throw TimeoutException('Network connection timed out');
},
);
} on HandshakeException {
if (rejectedCertificate != null) throw rejectedCertificate!;
rethrow;
@@ -408,7 +461,13 @@ class NetworkStorageService {
}
final HttpClientResponse response;
try {
response = await request.close().timeout(timeout);
response = await request.close().timeout(
timeout,
onTimeout: () {
request.abort();
throw TimeoutException('Network response timed out');
},
);
} on HandshakeException {
if (rejectedCertificate != null) throw rejectedCertificate!;
rethrow;
@@ -416,6 +475,7 @@ class NetworkStorageService {
if ([301, 302, 307, 308].contains(response.statusCode)) {
// Never replay a mutating request to an unselected path.
if (!['GET', 'HEAD', 'PROPFIND'].contains(method)) {
await _discardResponse(response);
throw const HttpException('Write destination redirected');
}
final location = response.headers.value(HttpHeaders.locationHeader);
@@ -429,6 +489,7 @@ class NetworkStorageService {
next.path != uri.path &&
next.path != '${uri.path}/') ||
redirects >= 3) {
await _discardResponse(response);
throw const HttpException('Unsupported server redirect');
}
await response.drain<void>().timeout(timeout);
@@ -479,44 +540,48 @@ class NetworkStorageService {
}
return prepareSmbEntries(files, path);
}
final client = _client();
try {
if (c.protocol == NetworkProtocol.http &&
NetworkEntry('', target.pathSegments.lastOrNull ?? '').audio) {
final response = await _request(
client,
c,
'GET',
target,
range: 'bytes=0-0',
);
if (response.statusCode != 200 && response.statusCode != 206) {
throw HttpException('Server returned ${response.statusCode}');
}
return [NetworkEntry(path, target.pathSegments.last)];
}
final client = _readClient(c);
if (c.protocol == NetworkProtocol.http &&
NetworkEntry('', target.pathSegments.lastOrNull ?? '').audio) {
final response = await _request(
client,
c,
c.protocol == NetworkProtocol.webdav ? 'PROPFIND' : 'GET',
'GET',
target,
dav: c.protocol == NetworkProtocol.webdav,
directoryListing: true,
range: 'bytes=0-0',
);
if (response.statusCode < 200 || response.statusCode >= 300) {
if (response.statusCode != 200 && response.statusCode != 206) {
await _discardResponse(response);
throw HttpException('Server returned ${response.statusCode}');
}
final bytes = BytesBuilder();
await for (final chunk in response.timeout(timeout)) {
if (bytes.length + chunk.length > 4 * 1024 * 1024) {
throw const HttpException('Directory listing is too large');
}
bytes.add(chunk);
// Do not download a whole track when a server ignores the range.
if (response.statusCode == 206 && response.contentLength == 1) {
await response.drain<void>().timeout(timeout);
} else {
await _discardResponse(response);
}
return await parseListingInBackground(c, path, bytes.takeBytes());
} finally {
client.close(force: true);
return [NetworkEntry(path, target.pathSegments.last)];
}
final response = await _request(
client,
c,
c.protocol == NetworkProtocol.webdav ? 'PROPFIND' : 'GET',
target,
dav: c.protocol == NetworkProtocol.webdav,
directoryListing: true,
);
if (response.statusCode < 200 || response.statusCode >= 300) {
await _discardResponse(response);
throw HttpException('Server returned ${response.statusCode}');
}
final bytes = BytesBuilder();
await for (final chunk in response.timeout(timeout)) {
if (bytes.length + chunk.length > 4 * 1024 * 1024) {
throw const HttpException('Directory listing is too large');
}
bytes.add(chunk);
}
return await parseListingInBackground(c, path, bytes.takeBytes());
}
/// Large SMB listings arrive from native I/O as plain stat objects. Transfer
@@ -689,7 +754,6 @@ class NetworkStorageService {
}
Future<void> _serve(HttpRequest request) async {
HttpClient? client;
NetworkSmbClient? smb;
try {
final route = _routes[request.uri.path.replaceFirst('/', '')];
@@ -735,7 +799,7 @@ class NetworkStorageService {
);
}
} else {
client = _client();
final client = _readClient(c);
final response = await _request(
client,
c,
@@ -758,6 +822,8 @@ class NetworkStorageService {
}
if (request.method == 'GET') {
await request.response.addStream(response.timeout(timeout));
} else {
await response.drain<void>().timeout(timeout);
}
}
} catch (_) {
@@ -765,7 +831,6 @@ class NetworkStorageService {
request.response.statusCode = HttpStatus.badGateway;
} catch (_) {}
} finally {
client?.close(force: true);
try {
await smb?.close();
} catch (_) {}
@@ -779,6 +844,10 @@ class NetworkStorageService {
final server = _server;
_server = null;
_routes.clear();
for (final client in _readClients.values) {
client.close(force: true);
}
_readClients.clear();
if (server != null) await (await server).close(force: true);
}
}
+104
View File
@@ -0,0 +1,104 @@
import 'dart:convert';
import 'dart:io';
import 'package:flutter_test/flutter_test.dart';
import 'package:spotiflac_android/services/network_storage_service.dart';
void main() {
TestWidgetsFlutterBinding.ensureInitialized();
HttpOverrides.global = null;
test(
'WebDAV listings and ranges reuse connections; saved credentials rotate the pool',
() async {
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
addTearDown(() => server.close(force: true));
final ports = <int>{};
final auth = <String?>[];
var active = 0;
var maximum = 0;
var failListing = false;
server.listen((request) async {
ports.add(request.connectionInfo!.remotePort);
auth.add(request.headers.value('authorization'));
await request.drain<void>();
if (request.method == 'PROPFIND') {
request.response.statusCode = failListing ? 500 : 207;
request.response.write(
failListing
? 'offline'
: '<d:multistatus xmlns:d="DAV:"><d:response><d:href>/music/song.flac</d:href>'
'<d:propstat><d:prop><d:getcontentlength>4</d:getcontentlength></d:prop>'
'<d:status>HTTP/1.1 200 OK</d:status></d:propstat></d:response></d:multistatus>',
);
} else {
expect(request.headers.value('range'), 'bytes=0-3');
active++;
if (active > maximum) maximum = active;
await Future<void>.delayed(const Duration(milliseconds: 15));
request.response.statusCode = 206;
request.response.headers.set('Content-Range', 'bytes 0-3/4');
request.response.contentLength = 4;
request.response.add([1, 2, 3, 4]);
active--;
}
await request.response.close();
});
final storage = NetworkStorageService(
read: () async => null,
write: (_) async {},
);
addTearDown(storage.dispose);
NetworkConnection connection(String password) => NetworkConnection(
id: 'pool',
name: 'Pool fixture',
protocol: NetworkProtocol.webdav,
address: 'http://127.0.0.1:${server.port}/music/',
username: 'user',
password: password,
);
final original = connection('first');
await storage.save(original);
final client = HttpClient();
addTearDown(() => client.close(force: true));
final proxy = await storage.resolve(
NetworkStorageService.source('pool', 'song.flac'),
);
Future<void> read() async {
final request = await client.getUrl(Uri.parse(proxy));
request.headers.set('Range', 'bytes=0-3');
final response = await request.close();
expect(response.statusCode, 206);
expect(response.headers.value('content-range'), 'bytes 0-3/4');
expect(await response.fold<List<int>>([], (a, b) => a..addAll(b)), [
1,
2,
3,
4,
]);
}
await storage.list(original);
await read();
await storage.list(original);
expect(ports, hasLength(1));
await Future.wait(List.generate(9, (_) => read()));
expect(maximum, greaterThan(1));
expect(maximum, lessThanOrEqualTo(6));
expect(ports.length, lessThanOrEqualTo(6));
expect(auth.toSet(), {
'Basic ${base64Encode(utf8.encode('user:first'))}',
});
failListing = true;
await expectLater(storage.list(original), throwsA(isA<HttpException>()));
failListing = false;
expect(await storage.list(original), hasLength(1));
final previous = ports.toSet();
final updated = connection('second');
await storage.save(updated);
await read();
expect(auth.last, 'Basic ${base64Encode(utf8.encode('user:second'))}');
expect(ports.difference(previous), hasLength(1));
},
);
}