Skip to content

Commit d0f8fe3

Browse files
chubes4chubes4
authored
Use bounded concurrent collector fetches (#824)
Use the validated concurrent transport for deterministic bounded page and asset collection while preserving limits, retry accounting, deadlines, and canonical output order. AI assistance: OpenAI gpt-5.6-sol via OpenCode implemented, tested, reconciled, and prepared this change. Chris Huber reviewed and remains responsible for it.
1 parent f62c06e commit d0f8fe3

2 files changed

Lines changed: 114 additions & 15 deletions

File tree

‎includes/class-static-site-importer-url-site-collector.php‎

Lines changed: 65 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,17 @@
1818
*/
1919
class Static_Site_Importer_URL_Site_Collector {
2020

21-
private const DEFAULT_MAX_PAGES = 20;
22-
private const DEFAULT_MAX_ASSETS = 200;
23-
private const DEFAULT_MAX_TOTAL_BYTES = 52428800;
24-
private const MAX_PAGES = 250;
25-
private const MAX_ASSETS = 2000;
26-
private const MAX_TOTAL_BYTES = 268435456;
27-
private const MAX_RESPONSE_BYTES = 10485760;
28-
private const MAX_SITEMAP_DOCUMENTS = 100;
29-
private const MAX_DISCOVERED_ROUTES = 5000;
30-
// The hardened socket transport is blocking, so one PHP invocation can safely admit one request at a time.
31-
private const SAME_ORIGIN_CONCURRENCY = 1;
32-
private const CROSS_ORIGIN_CONCURRENCY = 1;
21+
private const DEFAULT_MAX_PAGES = 20;
22+
private const DEFAULT_MAX_ASSETS = 200;
23+
private const DEFAULT_MAX_TOTAL_BYTES = 52428800;
24+
private const MAX_PAGES = 250;
25+
private const MAX_ASSETS = 2000;
26+
private const MAX_TOTAL_BYTES = 268435456;
27+
private const MAX_RESPONSE_BYTES = 10485760;
28+
private const MAX_SITEMAP_DOCUMENTS = 100;
29+
private const MAX_DISCOVERED_ROUTES = 5000;
30+
private const SAME_ORIGIN_CONCURRENCY = 2;
31+
private const CROSS_ORIGIN_CONCURRENCY = 4;
3332

3433
/**
3534
* Collect a public static site.
@@ -48,6 +47,7 @@ public static function collect( string $url, array $args = array(), ?callable $f
4847
$max_pages = min( self::MAX_PAGES, max( 1, (int) ( $args['max_pages'] ?? self::DEFAULT_MAX_PAGES ) ) );
4948
$max_assets = min( self::MAX_ASSETS, max( 0, (int) ( $args['max_assets'] ?? self::DEFAULT_MAX_ASSETS ) ) );
5049
$max_total_bytes = min( self::MAX_TOTAL_BYTES, max( 1, (int) ( $args['max_total_bytes'] ?? self::DEFAULT_MAX_TOTAL_BYTES ) ) );
50+
$use_many_fetcher = null === $fetcher;
5151
$fetcher = $fetcher ?? static fn ( string $resource_url, array $fetch_args ) => Static_Site_Importer_URL_Fetcher::fetch( $resource_url, $fetch_args );
5252
$fetcher = self::scheduled_fetcher( $fetcher, $args );
5353
$fetch_args = array_intersect_key( $args, array_flip( array( 'timeout' ) ) );
@@ -91,9 +91,10 @@ public static function collect( string $url, array $args = array(), ?callable $f
9191
}
9292
}
9393

94+
$page_fetcher = self::prefetched_fetcher( array_slice( $page_queue, 0, $max_pages ), array_merge( $fetch_args, array( 'content_types' => array( 'text/html', 'application/xhtml+xml' ) ) ), $fetcher, $use_many_fetcher, $args );
9495
while ( $page_queue && self::resource_count( $resources, 'html' ) < $max_pages ) {
9596
$page_url = array_shift( $page_queue );
96-
$response = $fetcher( $page_url, array_merge( $fetch_args, array( 'content_types' => array( 'text/html', 'application/xhtml+xml' ) ) ) );
97+
$response = $page_fetcher( $page_url, array_merge( $fetch_args, array( 'content_types' => array( 'text/html', 'application/xhtml+xml' ) ) ) );
9798
$response = self::without_cache_marker( $response );
9899
if ( is_wp_error( $response ) ) {
99100
if ( $page_url === $entry_url ) {
@@ -176,9 +177,10 @@ public static function collect( string $url, array $args = array(), ?callable $f
176177
}
177178
}
178179

180+
$asset_fetcher = self::prefetched_fetcher( array_slice( $asset_queue, 0, $max_assets ), array_merge( $fetch_args, array( 'content_types' => array() ) ), $fetcher, $use_many_fetcher, $args );
179181
while ( $asset_queue && self::resource_count( $resources, 'asset' ) < $max_assets ) {
180182
$asset_url = array_shift( $asset_queue );
181-
$response = $fetcher( $asset_url, array_merge( $fetch_args, array( 'content_types' => array() ) ) );
183+
$response = $asset_fetcher( $asset_url, array_merge( $fetch_args, array( 'content_types' => array() ) ) );
182184
$response = self::without_cache_marker( $response );
183185
if ( is_wp_error( $response ) ) {
184186
if ( $preserve_failed_assets ) {
@@ -872,6 +874,55 @@ private static function external_asset_samples( array $external_assets ): array
872874
return $samples;
873875
}
874876

877+
/** @return callable */
878+
private static function prefetched_fetcher( array $urls, array $fetch_args, callable $fetcher, bool $use_many_fetcher, array $args ): callable {
879+
if ( ! $use_many_fetcher || ! $urls ) {
880+
return $fetcher;
881+
}
882+
$responses = self::fetch_batch( $urls, $fetch_args, $args );
883+
return static fn ( string $url, array $request_args ) => $responses[ $url ] ?? $fetcher( $url, $request_args );
884+
}
885+
886+
/** @return array<string,array|WP_Error> */
887+
private static function fetch_batch( array $urls, array $fetch_args, array $args ): array {
888+
$attempts = min( 3, max( 1, (int) ( $args['fetch_attempts'] ?? 2 ) ) );
889+
$delay = min( 2000, max( 0, (int) ( $args['request_delay_ms'] ?? 0 ) ) );
890+
$pending = array_values( $urls );
891+
$responses = array();
892+
for ( $attempt = 0; $pending && $attempt < $attempts; $attempt++ ) {
893+
$requests = array();
894+
foreach ( $pending as $url ) {
895+
$requests[ $url ] = array(
896+
'url' => $url,
897+
'args' => $fetch_args,
898+
);
899+
}
900+
$many_args = array(
901+
'concurrency' => self::CROSS_ORIGIN_CONCURRENCY,
902+
'per_origin_concurrency' => self::SAME_ORIGIN_CONCURRENCY,
903+
);
904+
if ( isset( $args['_static_site_importer_fetch_deadline'] ) ) {
905+
$many_args['deadline'] = (float) $args['_static_site_importer_fetch_deadline'];
906+
}
907+
if ( isset( $args['_static_site_importer_fetch_clock'] ) && is_callable( $args['_static_site_importer_fetch_clock'] ) ) {
908+
$many_args['clock'] = $args['_static_site_importer_fetch_clock'];
909+
}
910+
if ( isset( $args['_static_site_importer_fetch_many_transport'] ) && is_array( $args['_static_site_importer_fetch_many_transport'] ) ) {
911+
$many_args['transport'] = $args['_static_site_importer_fetch_many_transport'];
912+
}
913+
$batch = Static_Site_Importer_URL_Fetcher::fetch_many( $requests, $many_args );
914+
$pending = array();
915+
foreach ( $batch as $url => $response ) {
916+
$responses[ $url ] = $response;
917+
if ( is_wp_error( $response ) && $attempt + 1 < $attempts ) {
918+
self::delay( $delay, $args );
919+
$pending[] = $url;
920+
}
921+
}
922+
}
923+
return $responses;
924+
}
925+
875926
/** @return callable */
876927
private static function scheduled_fetcher( callable $fetch_resource, array $args ): callable {
877928
$fetch_attempts = min( 3, max( 1, (int) ( $args['fetch_attempts'] ?? 2 ) ) );

‎tests/smoke-url-site-collector.php‎

Lines changed: 49 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -199,7 +199,55 @@ static function ( string $url, array $args ) use ( &$schedule_calls ): array {
199199
}
200200
);
201201
$assert( ! is_wp_error( $scheduled ) && 3 === count( $schedule_calls ) && array() === $schedule_delays, 'successful-uncached-fetches-have-no-default-pacing' );
202-
$assert( array( 'same_origin_concurrency' => 1, 'cross_origin_concurrency' => 1, 'retry_delay_ms' => 0 ) === ( $scheduled['source_metadata']['collection']['fetch_scheduling'] ?? null ), 'blocking-transport-scheduling-limits-are-explicit' );
202+
$assert( array( 'same_origin_concurrency' => 2, 'cross_origin_concurrency' => 4, 'retry_delay_ms' => 0 ) === ( $scheduled['source_metadata']['collection']['fetch_scheduling'] ?? null ), 'concurrent-transport-scheduling-limits-are-explicit' );
203+
204+
$concurrent_routes = array( 'http://1.1.1.1/', 'http://1.1.1.1/slow/', 'http://8.8.8.8/fast/', 'http://8.8.8.8/middle/' );
205+
$concurrent_active = array();
206+
$concurrent_origins = array();
207+
$concurrent_max_active = 0;
208+
$concurrent_max_origin = 0;
209+
$concurrent_starts = array();
210+
$concurrent_transport = array(
211+
'start' => static function ( array $target, array $options ) use ( &$concurrent_active, &$concurrent_origins, &$concurrent_max_active, &$concurrent_max_origin, &$concurrent_starts ) {
212+
unset( $options );
213+
$origin = $target['scheme'] . '://' . $target['host'] . ':' . $target['port'];
214+
$delay = array( '/' => 80000, '/slow/' => 60000, '/fast/' => 20000, '/middle/' => 40000 )[ $target['path'] ];
215+
$concurrent_active[] = $target['path'];
216+
$concurrent_origins[ $origin ] = ( $concurrent_origins[ $origin ] ?? 0 ) + 1;
217+
$concurrent_max_active = max( $concurrent_max_active, count( $concurrent_active ) );
218+
$concurrent_max_origin = max( $concurrent_max_origin, $concurrent_origins[ $origin ] );
219+
$concurrent_starts[] = $target['path'];
220+
return (object) array( 'target' => $target, 'origin' => $origin, 'due' => microtime( true ) + ( $delay / 1000000 ) );
221+
},
222+
'poll' => static function ( object $handle ) use ( &$concurrent_active, &$concurrent_origins ) {
223+
if ( microtime( true ) < $handle->due ) {
224+
return null;
225+
}
226+
$concurrent_active = array_values( array_filter( $concurrent_active, static fn ( string $path ): bool => $path !== $handle->target['path'] ) );
227+
--$concurrent_origins[ $handle->origin ];
228+
return array( 'status_code' => 200, 'headers' => array( 'content-type' => array( 'text/html' ) ), 'body' => '<main>' . $handle->target['path'] . '</main>' );
229+
},
230+
);
231+
$concurrent_started = microtime( true );
232+
$concurrent_result = Static_Site_Importer_URL_Site_Collector::collect(
233+
$concurrent_routes[0],
234+
array( 'max_pages' => 4, 'max_assets' => 0, 'fetch_attempts' => 1, '_route_set' => $concurrent_routes, '_static_site_importer_fetch_many_transport' => $concurrent_transport )
235+
);
236+
$concurrent_elapsed = microtime( true ) - $concurrent_started;
237+
$serial_started = microtime( true );
238+
$serial_result = Static_Site_Importer_URL_Site_Collector::collect(
239+
$concurrent_routes[0],
240+
array( 'max_pages' => 4, 'max_assets' => 0, 'fetch_attempts' => 1, '_route_set' => $concurrent_routes ),
241+
static function ( string $url, array $args ): array {
242+
unset( $args );
243+
usleep( array( '/' => 80000, '/slow/' => 60000, '/fast/' => 20000, '/middle/' => 40000 )[ (string) wp_parse_url( $url, PHP_URL_PATH ) ] );
244+
return array( 'body' => '<main>' . wp_parse_url( $url, PHP_URL_PATH ) . '</main>', 'metadata' => array( 'content_type' => 'text/html', 'final_url' => $url ) );
245+
}
246+
);
247+
$serial_elapsed = microtime( true ) - $serial_started;
248+
$assert( ! is_wp_error( $concurrent_result ) && 4 === $concurrent_max_active && 2 === $concurrent_max_origin && array( '/', '/slow/', '/fast/', '/middle/' ) === $concurrent_starts, 'collector-fetch-many-bounds-global-and-per-origin-admission' );
249+
$assert( ! is_wp_error( $concurrent_result ) && ! is_wp_error( $serial_result ) && ( $concurrent_result['source_metadata']['snapshot']['sha256'] ?? '' ) === ( $serial_result['source_metadata']['snapshot']['sha256'] ?? '' ) && ( $concurrent_result['source_metadata']['collection']['diagnostics'] ?? null ) === ( $serial_result['source_metadata']['collection']['diagnostics'] ?? null ), 'out-of-order-completions-preserve-serial-artifact-hash-and-diagnostics' );
250+
$assert( $concurrent_elapsed < $serial_elapsed * 0.7, 'bounded-concurrent-collection-materially-reduces-wall-clock-delay', sprintf( 'concurrent=%.3fs serial=%.3fs', $concurrent_elapsed, $serial_elapsed ) );
203251

204252
$retry_now = 0.0;
205253
$retry_calls = 0;

0 commit comments

Comments
 (0)