Skip to content

Commit 165c424

Browse files
chubes4chubes4
authored
fix: protect active batch parents during recovery (#2186)
1 parent 4cbb958 commit 165c424

4 files changed

Lines changed: 209 additions & 14 deletions

File tree

‎inc/Abilities/Engine/PipelineBatchScheduler.php‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,20 @@ public function fanOut(
121121
* @param int $parent_job_id The parent job ID.
122122
*/
123123
public function processChunk( int $parent_job_id ): void {
124+
$parent_job = $this->db_jobs->get_job( $parent_job_id );
125+
if ( ! $parent_job || JobStatus::PROCESSING !== ( $parent_job['status'] ?? '' ) ) {
126+
do_action(
127+
'datamachine_log',
128+
'warning',
129+
'Pipeline batch: skipped chunk because parent is no longer processing',
130+
array(
131+
'parent_job_id' => $parent_job_id,
132+
'parent_status' => $parent_job['status'] ?? 'missing',
133+
)
134+
);
135+
return;
136+
}
137+
124138
$result = BatchScheduler::processChunk(
125139
$parent_job_id,
126140
array( $this, 'createChildJobFromBatch' )

‎inc/Abilities/Job/RecoverStuckJobsAbility.php‎

Lines changed: 147 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -210,13 +210,13 @@ public function execute( array $input ): array {
210210
$job_id = (int) $job->job_id;
211211
$job_flow_id = (int) $job->flow_id;
212212

213-
if ( $this->hasActiveStepAction( $job_id, $timeout_hours ) ) {
213+
if ( $this->hasActiveSchedulerWork( $job_id, $engine_data, $timeout_hours ) ) {
214214
++$skipped;
215215
$jobs[] = array(
216216
'job_id' => $job_id,
217217
'flow_id' => $job_flow_id,
218218
'status' => 'skipped',
219-
'reason' => 'Pending or in-progress Action Scheduler step action exists',
219+
'reason' => 'Pending or in-progress scheduler work exists',
220220
);
221221
continue;
222222
}
@@ -436,7 +436,27 @@ private function getTerminalBackedInProgressActions( ?int $flow_id ): array {
436436
}
437437

438438
/**
439-
* Check whether Action Scheduler still owns executable work for a job.
439+
* Check whether scheduler or child work still owns executable work for a job.
440+
*
441+
* @param int $job_id Job ID.
442+
* @param array<string, mixed> $engine_data Job engine data.
443+
* @param int $timeout_hours Hours before in-progress actions are considered stale.
444+
* @return bool True when pending or fresh in-progress work exists.
445+
*/
446+
private function hasActiveSchedulerWork( int $job_id, array $engine_data, int $timeout_hours ): bool {
447+
if ( $this->hasActiveStepAction( $job_id, $timeout_hours ) ) {
448+
return true;
449+
}
450+
451+
if ( $this->hasActiveBatchWork( $job_id, $engine_data, $timeout_hours ) ) {
452+
return true;
453+
}
454+
455+
return false;
456+
}
457+
458+
/**
459+
* Check whether Action Scheduler still owns executable step work for a job.
440460
*
441461
* Pipeline jobs can stay in the `processing` state across multiple scheduled
442462
* steps. If a later `datamachine_execute_step` action is still pending or
@@ -500,35 +520,151 @@ private function hasActiveStepAction( int $job_id, int $timeout_hours ): bool {
500520
return false;
501521
}
502522

523+
/**
524+
* Check whether a pipeline batch parent still has scheduled chunk or child work.
525+
*
526+
* Batch parents wait on `datamachine_pipeline_batch_chunk` actions and child jobs,
527+
* not `datamachine_execute_step` actions. Treat both as live work so recovery
528+
* does not fail a parent while fan-out is still queued or children are active.
529+
*
530+
* @param int $parent_job_id Parent job ID.
531+
* @param array<string, mixed> $engine_data Parent engine data.
532+
* @param int $timeout_hours Hours before in-progress actions are considered stale.
533+
* @return bool True when batch chunk or child work is still active.
534+
*/
535+
private function hasActiveBatchWork( int $parent_job_id, array $engine_data, int $timeout_hours ): bool {
536+
if ( $parent_job_id <= 0 || empty( $engine_data['batch'] ) ) {
537+
return false;
538+
}
539+
540+
if ( $this->hasActiveActionForArg( 'datamachine_pipeline_batch_chunk', 'parent_job_id', $parent_job_id, $timeout_hours ) ) {
541+
return true;
542+
}
543+
544+
global $wpdb;
545+
$jobs_table = $wpdb->prefix . 'datamachine_jobs';
546+
547+
// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is generated from the WP prefix.
548+
$active_children = (int) $wpdb->get_var(
549+
$wpdb->prepare(
550+
"SELECT COUNT(*)
551+
FROM {$jobs_table}
552+
WHERE parent_job_id = %d
553+
AND status IN ( %s, %s )",
554+
$parent_job_id,
555+
'pending',
556+
'processing'
557+
)
558+
);
559+
// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared
560+
561+
return $active_children > 0;
562+
}
563+
564+
/**
565+
* Check for a pending or fresh in-progress Action Scheduler action by arg ID.
566+
*
567+
* @param string $hook Action hook.
568+
* @param string $arg_name Numeric argument name to match.
569+
* @param int $arg_id Expected numeric argument value.
570+
* @param int $timeout_hours Hours before in-progress actions are considered stale.
571+
* @return bool True when matching action is pending or freshly in-progress.
572+
*/
573+
private function hasActiveActionForArg( string $hook, string $arg_name, int $arg_id, int $timeout_hours ): bool {
574+
global $wpdb;
575+
576+
if ( $arg_id <= 0 ) {
577+
return false;
578+
}
579+
580+
$actions_table = $wpdb->prefix . 'actionscheduler_actions';
581+
$like_arg_id = '%"' . $wpdb->esc_like( $arg_name ) . '":' . $wpdb->esc_like( (string) $arg_id ) . '%';
582+
583+
// phpcs:disable WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- Table name is generated from the WP prefix.
584+
$actions = $wpdb->get_results(
585+
$wpdb->prepare(
586+
"SELECT action_id, args, status, scheduled_date_gmt, last_attempt_gmt
587+
FROM {$actions_table}
588+
WHERE hook = %s
589+
AND status IN ( %s, %s )
590+
AND args LIKE %s",
591+
$hook,
592+
'pending',
593+
'in-progress',
594+
$like_arg_id
595+
)
596+
);
597+
// phpcs:enable WordPress.DB.PreparedSQL.InterpolatedNotPrepared
598+
599+
$timeout_seconds = max( 1, $timeout_hours ) * HOUR_IN_SECONDS;
600+
$now_gmt = strtotime( current_time( 'mysql', true ) );
601+
602+
foreach ( $actions as $action ) {
603+
if ( $arg_id !== $this->extractActionArgInt( (string) ( $action->args ?? '' ), $arg_name ) ) {
604+
continue;
605+
}
606+
607+
if ( 'pending' === (string) $action->status ) {
608+
return true;
609+
}
610+
611+
$last_attempt = (string) ( $action->last_attempt_gmt ?? '' );
612+
$scheduled = (string) ( $action->scheduled_date_gmt ?? '' );
613+
$reference = $last_attempt && '0000-00-00 00:00:00' !== $last_attempt ? $last_attempt : $scheduled;
614+
$started_at = $reference ? strtotime( $reference ) : false;
615+
616+
if ( false === $started_at || false === $now_gmt ) {
617+
return true;
618+
}
619+
620+
if ( ( $now_gmt - $started_at ) < $timeout_seconds ) {
621+
return true;
622+
}
623+
}
624+
625+
return false;
626+
}
627+
503628
/**
504629
* Extract the Data Machine job ID from Action Scheduler args.
505630
*
506631
* @param string $args Action Scheduler args payload.
507632
* @return int Job ID, or 0 when unavailable.
508633
*/
509634
private function extractActionJobId( string $args ): int {
635+
return $this->extractActionArgInt( $args, 'job_id' );
636+
}
637+
638+
/**
639+
* Extract a numeric argument from an Action Scheduler args payload.
640+
*
641+
* @param string $args Action Scheduler args payload.
642+
* @param string $arg_name Argument name to extract.
643+
* @return int Argument value, or 0 when unavailable.
644+
*/
645+
private function extractActionArgInt( string $args, string $arg_name ): int {
510646
$decoded = json_decode( $args, true );
511647
if ( is_array( $decoded ) ) {
512-
if ( isset( $decoded['job_id'] ) && is_numeric( $decoded['job_id'] ) ) {
513-
return (int) $decoded['job_id'];
648+
if ( isset( $decoded[ $arg_name ] ) && is_numeric( $decoded[ $arg_name ] ) ) {
649+
return (int) $decoded[ $arg_name ];
514650
}
515651

516652
foreach ( $decoded as $value ) {
517-
if ( is_array( $value ) && isset( $value['job_id'] ) && is_numeric( $value['job_id'] ) ) {
518-
return (int) $value['job_id'];
653+
if ( is_array( $value ) && isset( $value[ $arg_name ] ) && is_numeric( $value[ $arg_name ] ) ) {
654+
return (int) $value[ $arg_name ];
519655
}
520656
}
521657
}
522658

523659
$unserialized = maybe_unserialize( $args );
524660
if ( is_array( $unserialized ) ) {
525-
if ( isset( $unserialized['job_id'] ) && is_numeric( $unserialized['job_id'] ) ) {
526-
return (int) $unserialized['job_id'];
661+
if ( isset( $unserialized[ $arg_name ] ) && is_numeric( $unserialized[ $arg_name ] ) ) {
662+
return (int) $unserialized[ $arg_name ];
527663
}
528664

529665
foreach ( $unserialized as $value ) {
530-
if ( is_array( $value ) && isset( $value['job_id'] ) && is_numeric( $value['job_id'] ) ) {
531-
return (int) $value['job_id'];
666+
if ( is_array( $value ) && isset( $value[ $arg_name ] ) && is_numeric( $value[ $arg_name ] ) ) {
667+
return (int) $value[ $arg_name ];
532668
}
533669
}
534670
}

‎tests/pipeline-batch-terminal-parent-guard-smoke.php‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
<?php
2+
/**
3+
* Static smoke test for terminal parent guard in pipeline batch chunks.
4+
*
5+
* Run with: php tests/pipeline-batch-terminal-parent-guard-smoke.php
6+
*
7+
* @package DataMachine\Tests
8+
*/
9+
10+
$failed = 0;
11+
$total = 0;
12+
13+
function assert_pipeline_batch_terminal_guard_smoke( string $name, bool $condition, string $detail = '' ): void {
14+
global $failed, $total;
15+
++$total;
16+
17+
if ( $condition ) {
18+
echo " [PASS] {$name}\n";
19+
return;
20+
}
21+
22+
echo " [FAIL] {$name}" . ( $detail ? " - {$detail}" : '' ) . "\n";
23+
++$failed;
24+
}
25+
26+
$source = file_get_contents( __DIR__ . '/../inc/Abilities/Engine/PipelineBatchScheduler.php' ) ?: '';
27+
$process_chunk = strstr( $source, 'public function processChunk' ) ?: '';
28+
29+
echo "Case 1: batch chunks refuse terminal parents\n";
30+
assert_pipeline_batch_terminal_guard_smoke( 'processChunk reads parent job before processing batch state', strpos( $process_chunk, '$parent_job = $this->db_jobs->get_job( $parent_job_id );' ) < strpos( $process_chunk, 'BatchScheduler::processChunk' ) );
31+
assert_pipeline_batch_terminal_guard_smoke( 'processChunk requires processing status', str_contains( $process_chunk, 'JobStatus::PROCESSING !== ( $parent_job[\'status\'] ?? \'\' )' ) );
32+
assert_pipeline_batch_terminal_guard_smoke( 'terminal parent guard returns before child creation', strpos( $process_chunk, 'return;' ) < strpos( $process_chunk, 'BatchScheduler::processChunk' ) );
33+
assert_pipeline_batch_terminal_guard_smoke( 'terminal parent guard logs skipped chunk', str_contains( $process_chunk, 'Pipeline batch: skipped chunk because parent is no longer processing' ) );
34+
35+
echo "\nPipeline batch terminal parent guard smoke complete: {$total} assertions, {$failed} failures.\n";
36+
if ( $failed > 0 ) {
37+
exit( 1 );
38+
}

‎tests/recover-stuck-active-action-guard-smoke.php‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,10 +26,11 @@ function assert_recover_stuck_guard_smoke( string $name, bool $condition, string
2626
$source = file_get_contents( __DIR__ . '/../inc/Abilities/Job/RecoverStuckJobsAbility.php' ) ?: '';
2727
$timeout_loop = strstr( $source, 'foreach ( $timed_out_jobs as $job )' ) ?: '';
2828

29-
echo "Case 1: timed-out recovery skips jobs with active step actions\n";
29+
echo "Case 1: timed-out recovery skips jobs with active scheduler work\n";
30+
assert_recover_stuck_guard_smoke( 'active scheduler work method exists', str_contains( $source, 'private function hasActiveSchedulerWork' ) );
3031
assert_recover_stuck_guard_smoke( 'active step guard method exists', str_contains( $source, 'private function hasActiveStepAction' ) );
31-
assert_recover_stuck_guard_smoke( 'timeout loop invokes active step guard before dry-run timeout', strpos( $timeout_loop, '$this->hasActiveStepAction( $job_id, $timeout_hours )' ) < strpos( $timeout_loop, 'if ( $dry_run )' ) );
32-
assert_recover_stuck_guard_smoke( 'guard records skipped status', str_contains( $source, "'status' => 'skipped'") && str_contains( $source, 'Pending or in-progress Action Scheduler step action exists' ) );
32+
assert_recover_stuck_guard_smoke( 'timeout loop invokes scheduler guard before dry-run timeout', strpos( $timeout_loop, '$this->hasActiveSchedulerWork( $job_id, $engine_data, $timeout_hours )' ) < strpos( $timeout_loop, 'if ( $dry_run )' ) );
33+
assert_recover_stuck_guard_smoke( 'guard records skipped status', str_contains( $source, "'status' => 'skipped'") && str_contains( $source, 'Pending or in-progress scheduler work exists' ) );
3334

3435
echo "Case 2: guard is limited to executable Data Machine step actions\n";
3536
assert_recover_stuck_guard_smoke( 'guard queries datamachine_execute_step', str_contains( $source, 'datamachine_execute_step' ) );
@@ -42,6 +43,12 @@ function assert_recover_stuck_guard_smoke( string $name, bool $condition, string
4243
assert_recover_stuck_guard_smoke( 'pending actions remain guarded unconditionally', str_contains( $source, 'if ( \'pending\' === (string) $action->status )' ) );
4344
assert_recover_stuck_guard_smoke( 'old in-progress actions can fall through', str_contains( $source, '( $now_gmt - $started_at ) < $timeout_seconds' ) );
4445

46+
echo "Case 4: batch parents guard chunk actions and active children\n";
47+
assert_recover_stuck_guard_smoke( 'active batch guard method exists', str_contains( $source, 'private function hasActiveBatchWork' ) );
48+
assert_recover_stuck_guard_smoke( 'batch guard checks pipeline chunk actions', str_contains( $source, 'datamachine_pipeline_batch_chunk' ) && str_contains( $source, 'parent_job_id' ) );
49+
assert_recover_stuck_guard_smoke( 'batch guard checks active children', str_contains( $source, 'WHERE parent_job_id = %d' ) && str_contains( $source, "status IN ( %s, %s )" ) );
50+
assert_recover_stuck_guard_smoke( 'generic action arg extractor supports parent ids', str_contains( $source, 'private function extractActionArgInt' ) && str_contains( $source, '$this->extractActionArgInt( (string) ( $action->args ?? \'\' ), $arg_name )' ) );
51+
4552
echo "\nRecover-stuck active action guard smoke complete: {$total} assertions, {$failed} failures.\n";
4653
if ( $failed > 0 ) {
4754
exit( 1 );

0 commit comments

Comments
 (0)