Skip to content

Commit 9e8a5e9

Browse files
chubes4chubes4
authored
Merge pull request #809 from Automattic/fix/url-batch-cooperative-continuation
Bound URL imports with cooperative continuations
2 parents ca30224 + 84ac5ff commit 9e8a5e9

2 files changed

Lines changed: 139 additions & 4 deletions

File tree

‎includes/class-static-site-importer-url-batch-import.php‎

Lines changed: 103 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,21 @@ public static function import( array $request, array $input, ?callable $fetcher
1212
$batch_pages = (int) ( $args['batch_pages'] ?? 0 );
1313
if ( $batch_pages < 1 ) {
1414
return new WP_Error('static_site_importer_invalid_batch_pages', 'batch_pages must be a positive integer.');
15+
}$max_effective_batches = null;
16+
if ( array_key_exists('max_effective_batches_per_invocation', $args) ) {
17+
$max_effective_batches = (int) $args['max_effective_batches_per_invocation'];
18+
if ( $max_effective_batches < 1 ) {
19+
return new WP_Error('static_site_importer_invalid_max_effective_batches_per_invocation', 'max_effective_batches_per_invocation must be a positive integer.');
20+
}
21+
}$clock = is_callable($args['_static_site_importer_clock'] ?? null) ? $args['_static_site_importer_clock'] : static fn(): float=>microtime(true);
22+
$deadline = null;
23+
$max_invocation_seconds = null;
24+
if ( array_key_exists('max_invocation_seconds', $args) ) {
25+
$max_invocation_seconds = (float) $args['max_invocation_seconds'];
26+
if ( $max_invocation_seconds <= 0 ) {
27+
return new WP_Error('static_site_importer_invalid_max_invocation_seconds', 'max_invocation_seconds must be a positive number.');
28+
}
29+
$deadline = (float) call_user_func($clock) + $max_invocation_seconds;
1530
}if ( ! array_key_exists('max_assets', $args) ) {
1631
$args['max_assets'] = 2000;
1732
}if ( ! array_key_exists('max_total_bytes', $args) ) {
@@ -50,7 +65,15 @@ public static function import( array $request, array $input, ?callable $fetcher
5065
});
5166
$cache->adopt_legacy(trailingslashit($work_dir) . 'url-response-cache-' . $identity);
5267
$cache->adopt_legacy($workspace->directory() . '/responses');
53-
$source_fetcher = $fetcher;
68+
$source_fetcher = $fetcher ?? static fn(string $resource_url,array $fetch_args)=>Static_Site_Importer_URL_Fetcher::fetch($resource_url, $fetch_args);
69+
if ( null !== $deadline ) {
70+
$source = $source_fetcher;
71+
$source_fetcher = static function(string $resource_url,array $fetch_args)use($source,$clock,$deadline) {
72+
if ( self::deadline_reached($deadline, $clock) ) {
73+
return new WP_Error('static_site_importer_invocation_deadline_exceeded', 'The URL batch invocation deadline was reached before starting a network fetch.');
74+
}return $source($resource_url, $fetch_args);
75+
};
76+
}
5477
$fetcher = self::cached_fetcher($cache, $source_fetcher);
5578
$run_manifest = new Static_Site_Importer_Artifact_Run_Manifest($manifest_path, $identity, 'static-site-importer/url-site-batch-run/v1', $contract);
5679
$manifest = $run_manifest->load();
@@ -103,7 +126,15 @@ public static function import( array $request, array $input, ?callable $fetcher
103126
return $manifest['final_result'];
104127
}$importer = $importer ?? static fn(array $artifact,array $import_args)=>Static_Site_Importer_Theme_Generator::import_website_artifact($artifact, $import_args);
105128
$cursor = Static_Site_Importer_Artifact_Batch_Cursor::hydrate($manifest['batches']);
129+
$effective_batches = 0;
106130
while ( null !== ( $index = Static_Site_Importer_Artifact_Batch_Cursor::next($cursor) ) ) {
131+
if ( null !== $max_effective_batches && $effective_batches >= $max_effective_batches ) {
132+
$manifest['batches'] = self::legacy_batches($cursor);
133+
self::checkpoint_cache($manifest, $cache);
134+
if ( is_wp_error($run_manifest->save($manifest)) ) {
135+
return $run_manifest->save($manifest);
136+
}return self::continuation_result($manifest, $manifest_path, $index, $effective_batches, $max_effective_batches);
137+
}
107138
$batch = $cursor[ $index ];
108139
$routes = array_values(array_intersect_key($manifest['routes'], array_flip($batch['units'])));
109140
$batch_entry = in_array($url, $routes, true) ? $url : ( $routes[0] ?? $url );
@@ -119,6 +150,14 @@ public static function import( array $request, array $input, ?callable $fetcher
119150
$collect_args['asset_failure_policy'] = count($routes) > 1 ? 'preserve_failed_external_assets' : 'preserve_external';
120151
$runtime = Static_Site_Importer_URL_Site_Collector::collect($batch_entry, $collect_args, $fetcher);
121152
if ( is_wp_error($runtime) ) {
153+
if ( self::deadline_error($runtime) ) {
154+
$manifest['batches'] = self::legacy_batches($cursor);
155+
self::checkpoint_cache($manifest, $cache);
156+
$write = $run_manifest->save($manifest);
157+
if ( is_wp_error($write) ) {
158+
return $write;
159+
}return self::continuation_result($manifest, $manifest_path, $index, $effective_batches, $max_effective_batches, $max_invocation_seconds, 'deadline_exhausted');
160+
}
122161
if ( count($routes) > 1 && self::splittable_collection_error($runtime) ) {
123162
$cursor = Static_Site_Importer_Artifact_Batch_Cursor::split($cursor, $index);
124163
$manifest['batches'] = self::legacy_batches($cursor);
@@ -128,8 +167,14 @@ public static function import( array $request, array $input, ?callable $fetcher
128167
'parent_batch' => $batch['batch_id'],
129168
'children' => array_column(array_slice($cursor, $index, 2), 'batch_id'),
130169
);
131-
$run_manifest->save($manifest);
132-
return self::import($request, $input, $source_fetcher, $importer);
170+
$write = $run_manifest->save($manifest);
171+
if ( is_wp_error($write) ) {
172+
return $write;
173+
}
174+
if ( null !== $max_effective_batches ) {
175+
return self::continuation_result($manifest, $manifest_path, $index, $effective_batches, $max_effective_batches, $max_invocation_seconds, 'batch_subdivided');
176+
}
177+
continue;
133178
}return self::failed($run_manifest, $workspace, $manifest, $cursor, $index, $runtime, $cache);
134179
}$write = $workspace->publish_json($cache_name, $runtime);
135180
if ( is_wp_error($write) ) {
@@ -140,6 +185,12 @@ public static function import( array $request, array $input, ?callable $fetcher
140185
$manifest['batches'] = self::legacy_batches($cursor);
141186
if ( is_wp_error($run_manifest->save($manifest)) ) {
142187
return $run_manifest->save($manifest);
188+
}if ( null !== $deadline && self::deadline_reached($deadline, $clock) ) {
189+
self::checkpoint_cache($manifest, $cache);
190+
$write = $run_manifest->save($manifest);
191+
if ( is_wp_error($write) ) {
192+
return $write;
193+
}return self::continuation_result($manifest, $manifest_path, $index, $effective_batches, $max_effective_batches, $max_invocation_seconds, 'deadline_exhausted');
143194
}$import_args = Static_Site_Importer_URL_Import_Runtime::batch_import_args($input, $runtime);
144195
$import_args['activate'] = $index === array_key_last($cursor) && ! empty($input['activate']);
145196
$import_args['batch_import'] = true;
@@ -157,7 +208,8 @@ public static function import( array $request, array $input, ?callable $fetcher
157208
}$workspace->delete($cache_name);
158209
if ( is_file($old_cache) ) {
159210
unlink($old_cache);
160-
}$final = $result;
211+
}$effective_batches++;
212+
$final = $result;
161213
unset($result, $runtime, $raw);}
162214
$manifest['batches'] = self::legacy_batches($cursor);
163215
$aggregate = self::aggregate_result($manifest, $manifest_path, $final ?? array());
@@ -325,6 +377,15 @@ private static function splittable_collection_error(WP_Error $error): bool {$dat
325377
return true;
326378
}
327379
}return false;}
380+
private static function deadline_error(WP_Error $error): bool {if ( 'static_site_importer_invocation_deadline_exceeded' === $error->get_error_code() ) {
381+
return true;
382+
}$data = $error->get_error_data();
383+
foreach ( is_array($data) ? ( $data['collection']['failures'] ?? array() ) : array() as $failure ) {
384+
if ( 'static_site_importer_invocation_deadline_exceeded' === ( $failure['code'] ?? '' ) ) {
385+
return true;
386+
}
387+
}return false;}
388+
private static function deadline_reached(float $deadline,callable $clock): bool {return (float) call_user_func($clock) >= $deadline;}
328389
private static function result_evidence(array $result,array $runtime): array {return array(
329390
'theme_slug' => $result['theme_slug'] ?? '',
330391
'snapshot_sha256' => $runtime['source_metadata']['snapshot']['sha256'] ?? '',
@@ -385,6 +446,44 @@ private static function aggregate_result(array $manifest,string $path,array $ter
385446
'batch_materialization' => $manifest['batches'],
386447
'terminal_batch_result' => $terminal,
387448
);}
449+
private static function continuation_result(array $manifest,string $path,int $index,int $effective_batches,?int $max_effective_batches = null,?float $max_invocation_seconds = null,string $reason = 'effective_batch_limit'): array {$next = $manifest['batches'][ $index ] ?? array();
450+
$next_work = array_filter(array(
451+
'index' => $next['index'] ?? $index,
452+
'batch_id' => $next['batch_id'] ?? '',
453+
'route_indexes' => $next['route_indexes'] ?? array(),
454+
'effective_batch_size' => $next['effective_batch_size'] ?? null,
455+
), static fn($value): bool=>null !== $value);
456+
$completed_batches = count(array_filter($manifest['batches'], static fn(array $batch): bool=>'completed' === $batch['state']));
457+
$completed_routes = array_sum(array_column($manifest['batches'], 'completed_routes'));
458+
return array(
459+
'success' => true,
460+
'continuation' => true,
461+
'continuation_reason' => $reason,
462+
'import_report_summary' => array(
463+
'status' => 'continuing',
464+
'scope' => 'url_site_batch_run',
465+
'total_routes' => $manifest['total_routes'],
466+
'completed_routes' => $completed_routes,
467+
'total_batches' => count($manifest['batches']),
468+
'completed_batches' => $completed_batches,
469+
),
470+
'url_batch_run' => array(
471+
'status' => 'continuing',
472+
'run_manifest' => $path,
473+
'fetch_cache' => $manifest['fetch_cache'] ?? array(),
474+
'per_batch_limits' => $manifest['per_batch_limits'] ?? array(),
475+
'total_routes' => $manifest['total_routes'],
476+
'completed_routes' => $completed_routes,
477+
'total_batches' => count($manifest['batches']),
478+
'completed_batches' => $completed_batches,
479+
'effective_batches_processed' => $effective_batches,
480+
'max_effective_batches_per_invocation' => $max_effective_batches,
481+
'max_invocation_seconds' => $max_invocation_seconds,
482+
'continuation_reason' => $reason,
483+
'next_work' => $next_work,
484+
),
485+
'batch_materialization' => $manifest['batches'],
486+
);}
388487
private static function contract(string $url,array $input,array $args,int $batch_pages): array {foreach ( array_keys($args)as$key ) {
389488
if ( str_starts_with( (string) $key, '_static_site_importer_') ) {
390489
unset($args[ $key ]);

0 commit comments

Comments
 (0)