Skip to content

Use bounded concurrent collector fetches #824

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chubes4 merged 6 commits into from
Aug 3, 2026
Merged

Use bounded concurrent collector fetches #824

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8415cce
Add bounded concurrent URL transport
chubes4 Aug 3, 2026
e2703d5
Clean concurrent transport formatting
chubes4 Aug 3, 2026
c5b5f59
Bound URL fetch retry pacing
chubes4 Aug 3, 2026
bc7b97a
Fix collector lint findings
chubes4 Aug 3, 2026
4a06610
Use bounded concurrent collector fetches
chubes4 Aug 3, 2026
9232df3
Merge origin/main into feat/814-concurrent-collector
chubes4 Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 65 additions & 14 deletions includes/class-static-site-importer-url-site-collector.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,17 @@
*/
class Static_Site_Importer_URL_Site_Collector {

private const DEFAULT_MAX_PAGES = 20;
private const DEFAULT_MAX_ASSETS = 200;
private const DEFAULT_MAX_TOTAL_BYTES = 52428800;
private const MAX_PAGES = 250;
private const MAX_ASSETS = 2000;
private const MAX_TOTAL_BYTES = 268435456;
private const MAX_RESPONSE_BYTES = 10485760;
private const MAX_SITEMAP_DOCUMENTS = 100;
private const MAX_DISCOVERED_ROUTES = 5000;
// The hardened socket transport is blocking, so one PHP invocation can safely admit one request at a time.
private const SAME_ORIGIN_CONCURRENCY = 1;
private const CROSS_ORIGIN_CONCURRENCY = 1;
private const DEFAULT_MAX_PAGES = 20;
private const DEFAULT_MAX_ASSETS = 200;
private const DEFAULT_MAX_TOTAL_BYTES = 52428800;
private const MAX_PAGES = 250;
private const MAX_ASSETS = 2000;
private const MAX_TOTAL_BYTES = 268435456;
private const MAX_RESPONSE_BYTES = 10485760;
private const MAX_SITEMAP_DOCUMENTS = 100;
private const MAX_DISCOVERED_ROUTES = 5000;
private const SAME_ORIGIN_CONCURRENCY = 2;
private const CROSS_ORIGIN_CONCURRENCY = 4;

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

$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 );
while ( $page_queue && self::resource_count( $resources, 'html' ) < $max_pages ) {
$page_url = array_shift( $page_queue );
$response = $fetcher( $page_url, array_merge( $fetch_args, array( 'content_types' => array( 'text/html', 'application/xhtml+xml' ) ) ) );
$response = $page_fetcher( $page_url, array_merge( $fetch_args, array( 'content_types' => array( 'text/html', 'application/xhtml+xml' ) ) ) );
$response = self::without_cache_marker( $response );
if ( is_wp_error( $response ) ) {
if ( $page_url === $entry_url ) {
Expand Down Expand Up @@ -176,9 +177,10 @@ public static function collect( string $url, array $args = array(), ?callable $f
}
}

$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 );
while ( $asset_queue && self::resource_count( $resources, 'asset' ) < $max_assets ) {
$asset_url = array_shift( $asset_queue );
$response = $fetcher( $asset_url, array_merge( $fetch_args, array( 'content_types' => array() ) ) );
$response = $asset_fetcher( $asset_url, array_merge( $fetch_args, array( 'content_types' => array() ) ) );
$response = self::without_cache_marker( $response );
if ( is_wp_error( $response ) ) {
if ( $preserve_failed_assets ) {
Expand Down Expand Up @@ -872,6 +874,55 @@ private static function external_asset_samples( array $external_assets ): array
return $samples;
}

/** @return callable */
private static function prefetched_fetcher( array $urls, array $fetch_args, callable $fetcher, bool $use_many_fetcher, array $args ): callable {
if ( ! $use_many_fetcher || ! $urls ) {
return $fetcher;
}
$responses = self::fetch_batch( $urls, $fetch_args, $args );
return static fn ( string $url, array $request_args ) => $responses[ $url ] ?? $fetcher( $url, $request_args );
}

/** @return array<string,array|WP_Error> */
private static function fetch_batch( array $urls, array $fetch_args, array $args ): array {
$attempts = min( 3, max( 1, (int) ( $args['fetch_attempts'] ?? 2 ) ) );
$delay = min( 2000, max( 0, (int) ( $args['request_delay_ms'] ?? 0 ) ) );
$pending = array_values( $urls );
$responses = array();
for ( $attempt = 0; $pending && $attempt < $attempts; $attempt++ ) {
$requests = array();
foreach ( $pending as $url ) {
$requests[ $url ] = array(
'url' => $url,
'args' => $fetch_args,
);
}
$many_args = array(
'concurrency' => self::CROSS_ORIGIN_CONCURRENCY,
'per_origin_concurrency' => self::SAME_ORIGIN_CONCURRENCY,
);
if ( isset( $args['_static_site_importer_fetch_deadline'] ) ) {
$many_args['deadline'] = (float) $args['_static_site_importer_fetch_deadline'];
}
if ( isset( $args['_static_site_importer_fetch_clock'] ) && is_callable( $args['_static_site_importer_fetch_clock'] ) ) {
$many_args['clock'] = $args['_static_site_importer_fetch_clock'];
}
if ( isset( $args['_static_site_importer_fetch_many_transport'] ) && is_array( $args['_static_site_importer_fetch_many_transport'] ) ) {
$many_args['transport'] = $args['_static_site_importer_fetch_many_transport'];
}
$batch = Static_Site_Importer_URL_Fetcher::fetch_many( $requests, $many_args );
$pending = array();
foreach ( $batch as $url => $response ) {
$responses[ $url ] = $response;
if ( is_wp_error( $response ) && $attempt + 1 < $attempts ) {
self::delay( $delay, $args );
$pending[] = $url;
}
}
}
return $responses;
}

/** @return callable */
private static function scheduled_fetcher( callable $fetch_resource, array $args ): callable {
$fetch_attempts = min( 3, max( 1, (int) ( $args['fetch_attempts'] ?? 2 ) ) );
Expand Down
50 changes: 49 additions & 1 deletion tests/smoke-url-site-collector.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,55 @@ static function ( string $url, array $args ) use ( &$schedule_calls ): array {
}
);
$assert( ! is_wp_error( $scheduled ) && 3 === count( $schedule_calls ) && array() === $schedule_delays, 'successful-uncached-fetches-have-no-default-pacing' );
$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' );
$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' );

$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/' );
$concurrent_active = array();
$concurrent_origins = array();
$concurrent_max_active = 0;
$concurrent_max_origin = 0;
$concurrent_starts = array();
$concurrent_transport = array(
'start' => static function ( array $target, array $options ) use ( &$concurrent_active, &$concurrent_origins, &$concurrent_max_active, &$concurrent_max_origin, &$concurrent_starts ) {
unset( $options );
$origin = $target['scheme'] . '://' . $target['host'] . ':' . $target['port'];
$delay = array( '/' => 80000, '/slow/' => 60000, '/fast/' => 20000, '/middle/' => 40000 )[ $target['path'] ];
$concurrent_active[] = $target['path'];
$concurrent_origins[ $origin ] = ( $concurrent_origins[ $origin ] ?? 0 ) + 1;
$concurrent_max_active = max( $concurrent_max_active, count( $concurrent_active ) );
$concurrent_max_origin = max( $concurrent_max_origin, $concurrent_origins[ $origin ] );
$concurrent_starts[] = $target['path'];
return (object) array( 'target' => $target, 'origin' => $origin, 'due' => microtime( true ) + ( $delay / 1000000 ) );
},
'poll' => static function ( object $handle ) use ( &$concurrent_active, &$concurrent_origins ) {
if ( microtime( true ) < $handle->due ) {
return null;
}
$concurrent_active = array_values( array_filter( $concurrent_active, static fn ( string $path ): bool => $path !== $handle->target['path'] ) );
--$concurrent_origins[ $handle->origin ];
return array( 'status_code' => 200, 'headers' => array( 'content-type' => array( 'text/html' ) ), 'body' => '<main>' . $handle->target['path'] . '</main>' );
},
);
$concurrent_started = microtime( true );
$concurrent_result = Static_Site_Importer_URL_Site_Collector::collect(
$concurrent_routes[0],
array( 'max_pages' => 4, 'max_assets' => 0, 'fetch_attempts' => 1, '_route_set' => $concurrent_routes, '_static_site_importer_fetch_many_transport' => $concurrent_transport )
);
$concurrent_elapsed = microtime( true ) - $concurrent_started;
$serial_started = microtime( true );
$serial_result = Static_Site_Importer_URL_Site_Collector::collect(
$concurrent_routes[0],
array( 'max_pages' => 4, 'max_assets' => 0, 'fetch_attempts' => 1, '_route_set' => $concurrent_routes ),
static function ( string $url, array $args ): array {
unset( $args );
usleep( array( '/' => 80000, '/slow/' => 60000, '/fast/' => 20000, '/middle/' => 40000 )[ (string) wp_parse_url( $url, PHP_URL_PATH ) ] );
return array( 'body' => '<main>' . wp_parse_url( $url, PHP_URL_PATH ) . '</main>', 'metadata' => array( 'content_type' => 'text/html', 'final_url' => $url ) );
}
);
$serial_elapsed = microtime( true ) - $serial_started;
$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' );
$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' );
$assert( $concurrent_elapsed < $serial_elapsed * 0.7, 'bounded-concurrent-collection-materially-reduces-wall-clock-delay', sprintf( 'concurrent=%.3fs serial=%.3fs', $concurrent_elapsed, $serial_elapsed ) );

$retry_now = 0.0;
$retry_calls = 0;
Expand Down
Loading