Skip to content

fix: stop idempotent job insert storms #3059

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 1 commit into from
Aug 6, 2026
Merged

fix: stop idempotent job insert storms #3059

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
137 changes: 104 additions & 33 deletions inc/Core/Database/Jobs/Jobs.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,12 +33,15 @@ class Jobs extends BaseRepository {

const TABLE_NAME = 'datamachine_jobs';

private const TERMINAL_ACCOUNTING_METRICS = 0;
private const TERMINAL_ACCOUNTING_CORE = 1;
private const TERMINAL_ACCOUNTING_LIFECYCLE = 2;
private const TERMINAL_ACCOUNTING_NOTIFY = 3;
public const TERMINAL_ACCOUNTING_COMPLETE = 4;
private const RECOVERY_LEASE_TTL = 300;
private const TERMINAL_ACCOUNTING_METRICS = 0;
private const TERMINAL_ACCOUNTING_CORE = 1;
private const TERMINAL_ACCOUNTING_LIFECYCLE = 2;
private const TERMINAL_ACCOUNTING_NOTIFY = 3;
public const TERMINAL_ACCOUNTING_COMPLETE = 4;
private const RECOVERY_LEASE_TTL = 300;
private const IDEMPOTENT_INSERT_MAX_ATTEMPTS = 3;
private const IDEMPOTENT_INSERT_RETRY_US = 10000;
private const INSERT_FAILURE_LOG_TTL = 300;

/** Job currently owning the connection-wide terminal transaction. */
private static ?int $terminalizing_job = null;
Expand Down Expand Up @@ -191,37 +194,38 @@ public function create_or_get_job( array $job_data ): array|false {
return false;
}

$inserted = $this->insert_prepared_job( $prepared );
if ( false === $inserted ) {
$db_error = '';
$attempts = 0;
while ( $attempts < self::IDEMPOTENT_INSERT_MAX_ATTEMPTS ) {
++$attempts;
$inserted = $this->insert_prepared_job( $prepared );
if ( false !== $inserted ) {
break;
}

// Preserve the write error before the recovery read resets wpdb::last_error.
$db_error = (string) $this->wpdb->last_error;
$existing = $this->get_job_by_idempotency_key( $idempotency_key );
if ( null !== $existing ) {
( new RunLifecycleStore( $this ) )->mark_job_created(
(int) $existing['job_id'],
array(
'run_type' => $existing['source'] ?? 'job',
'status' => $existing['status'] ?? JobStatus::PENDING,
)
);
return $this->existing_idempotent_job_result( $existing );
}

return array(
'job_id' => (int) $existing['job_id'],
'created' => false,
'already_exists' => true,
'job' => $existing,
);
if ( $this->is_duplicate_database_error( $db_error ) ) {
$existing = $this->wait_for_idempotent_race_winner( $idempotency_key );
if ( null !== $existing ) {
return $this->existing_idempotent_job_result( $existing );
}
break;
}

do_action(
'datamachine_log',
'error',
'Failed to insert idempotent job',
array(
'pipeline_id' => $prepared['pipeline_id'],
'flow_id' => $prepared['flow_id'],
'idempotency_key' => $idempotency_key,
'db_error' => $this->wpdb->last_error,
)
);
if ( ! $this->is_retryable_database_error( $db_error ) || $attempts >= self::IDEMPOTENT_INSERT_MAX_ATTEMPTS ) {
break;
}
usleep( self::IDEMPOTENT_INSERT_RETRY_US * $attempts );
}

if ( false === $inserted ) {
$this->report_idempotent_insert_failure( $prepared, $idempotency_key, $db_error, $attempts );
return false;
}

Expand Down Expand Up @@ -318,7 +322,7 @@ private function prepare_job_insert( array $job_data ): array|false {
$default_source = $is_contextless ? 'direct' : ( $is_direct_execution ? 'direct' : 'pipeline' );
$source = sanitize_key( $job_data['source'] ?? $default_source );

$label = isset( $job_data['label'] ) ? sanitize_text_field( $job_data['label'] ) : null;
$label = isset( $job_data['label'] ) ? mb_substr( sanitize_text_field( $job_data['label'] ), 0, 255, 'UTF-8' ) : null;

$parent_job_id = isset( $job_data['parent_job_id'] ) ? absint( $job_data['parent_job_id'] ) : 0;
$user_id = isset( $job_data['user_id'] ) ? absint( $job_data['user_id'] ) : 0;
Expand Down Expand Up @@ -684,6 +688,73 @@ private function insert_prepared_job( array $prepared ): int|false {
return $this->wpdb->insert( $this->table_name, $prepared['data'], $prepared['format'] );
}

/** Build the canonical response for a competing idempotent insert winner. */
private function existing_idempotent_job_result( array $existing ): array {
( new RunLifecycleStore( $this ) )->mark_job_created(
(int) $existing['job_id'],
array(
'run_type' => $existing['source'] ?? 'job',
'status' => $existing['status'] ?? JobStatus::PENDING,
)
);

return array(
'job_id' => (int) $existing['job_id'],
'created' => false,
'already_exists' => true,
'job' => $existing,
);
}

/** Give a committed duplicate-race winner a bounded visibility window. */
private function wait_for_idempotent_race_winner( string $idempotency_key ): ?array {
for ( $attempt = 1; $attempt < self::IDEMPOTENT_INSERT_MAX_ATTEMPTS; ++$attempt ) {
usleep( self::IDEMPOTENT_INSERT_RETRY_US * $attempt );
$existing = $this->get_job_by_idempotency_key( $idempotency_key );
if ( null !== $existing ) {
return $existing;
}
}

return null;
}

/** Whether an insert error proves another writer won the unique-key race. */
private function is_duplicate_database_error( string $db_error ): bool {
return str_contains( $db_error, '1062' ) || false !== stripos( $db_error, 'Duplicate entry' );
}

/** Retry only bounded transaction-contention failures. */
private function is_retryable_database_error( string $db_error ): bool {
return str_contains( $db_error, '1205' )
|| str_contains( $db_error, '1213' )
|| false !== stripos( $db_error, 'Lock wait timeout' )
|| false !== stripos( $db_error, 'Deadlock found' );
}

/** Emit safe, rate-limited diagnostics while keeping a per-failure signal. */
private function report_idempotent_insert_failure( array $prepared, string $idempotency_key, string $db_error, int $attempts ): void {
$error_type = $this->is_duplicate_database_error( $db_error ) ? 'duplicate_without_winner' : ( $this->is_retryable_database_error( $db_error ) ? 'contention_exhausted' : 'non_retryable' );
$key_type = str_contains( $idempotency_key, ':' ) ? strstr( $idempotency_key, ':', true ) : 'untyped';
$context = array(
'pipeline_id' => $prepared['pipeline_id'],
'flow_id' => $prepared['flow_id'],
'idempotency_key_type' => sanitize_key( $key_type ),
'idempotency_key_hash' => hash( 'sha256', $idempotency_key ),
'db_error' => mb_substr( sanitize_text_field( $db_error ), 0, 500, 'UTF-8' ),
'error_type' => $error_type,
'attempts' => $attempts,
'log_throttle_seconds' => self::INSERT_FAILURE_LOG_TTL,
);

do_action( 'datamachine_idempotent_job_insert_failed', $context );

$throttle_key = 'insert_failure_' . hash( 'sha256', $error_type . '|' . $prepared['pipeline_id'] . '|' . $prepared['flow_id'] . '|' . $idempotency_key );
if ( wp_cache_add( $throttle_key, 1, 'datamachine_job_insert_failures', self::INSERT_FAILURE_LOG_TTL ) ) {
do_action( 'datamachine_log', 'error', 'Failed to insert idempotent job', $context );
}
}

/**
* Normalize a job idempotency key for storage and lookup.
*
Expand Down
198 changes: 198 additions & 0 deletions tests/jobs-idempotent-insert-recovery-smoke.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,198 @@
<?php
/**
* Regression coverage for bounded idempotent job insertion recovery.
*
* Run with: php tests/jobs-idempotent-insert-recovery-smoke.php
*/

namespace DataMachine\Core {
class JobStatus {
public const PENDING = 'pending';
}

class RunLifecycleStore {
public function __construct( mixed $jobs = null ) {}

public function mark_job_created( int $job_id, array $seed = array() ): void {}
}
}

namespace {
define( 'ABSPATH', __DIR__ . '/' );
define( 'ARRAY_A', 'ARRAY_A' );
define( 'OBJECT', 'OBJECT' );

$GLOBALS['job_insert_test_actions'] = array();
$GLOBALS['job_insert_test_cache'] = array();

function absint( mixed $value ): int {
return abs( (int) $value );
}

function sanitize_key( mixed $value ): string {
return preg_replace( '/[^a-z0-9_\-]/', '', strtolower( (string) $value ) );
}

function sanitize_text_field( mixed $value ): string {
return trim( strip_tags( (string) $value ) );
}

function wp_json_encode( mixed $value ): string|false {
return json_encode( $value );
}

function current_time( string $type, bool $gmt = false ): string {
return '2026-08-06 15:00:00';
}

function do_action( string $hook, mixed ...$args ): void {
$GLOBALS['job_insert_test_actions'][ $hook ][] = $args;
}

function wp_cache_add( string $key, mixed $value, string $group = '', int $expire = 0 ): bool {
if ( isset( $GLOBALS['job_insert_test_cache'][ $group ][ $key ] ) ) {
return false;
}
$GLOBALS['job_insert_test_cache'][ $group ][ $key ] = $value;
return true;
}

class wpdb {
public string $prefix = 'wp_';
public int $insert_id = 0;
public string $last_error = '';
}

class Jobs_Idempotent_Insert_Test_Wpdb extends wpdb {
public string $mode = 'success';
public int $insert_calls = 0;
public int $lookup_calls = 0;
/** @var array<int,array<string,mixed>> */
public array $rows = array();

public function insert( mixed $table, array $data, array $format ): int|false {
++$this->insert_calls;
if ( 'duplicate_race' === $this->mode ) {
$this->last_error = "Duplicate entry 'redacted' for key 'idx_idempotency_key' (1062)";
return false;
}
if ( 'contention' === $this->mode ) {
$this->last_error = 'Deadlock found when trying to get lock; try restarting transaction (1213)';
return false;
}
if ( 'contention_then_success' === $this->mode && $this->insert_calls < 3 ) {
$this->last_error = 'Lock wait timeout exceeded; try restarting transaction (1205)';
return false;
}
if ( 'non_retryable' === $this->mode ) {
$this->last_error = "Data too long for column 'operation_step_id' at row 1";
return false;
}

$this->last_error = '';
$this->insert_id = count( $this->rows ) + 1;
$this->rows[ $this->insert_id ] = array_merge(
array(
'job_id' => $this->insert_id,
'created_at' => '2026-08-06 15:00:00',
),
$data
);
return 1;
}

public function prepare( string $query, mixed ...$args ): array {
return array( $query, $args );
}

public function get_row( mixed $query, mixed $output = OBJECT ): ?array {
$args = is_array( $query ) ? $query[1] : array();
if ( str_contains( (string) $query[0], 'idempotency_key' ) ) {
++$this->lookup_calls;
$key = (string) end( $args );
if ( 'duplicate_race' === $this->mode && $this->lookup_calls >= 3 ) {
return array(
'job_id' => 77,
'idempotency_key' => $key,
'source' => 'pipeline',
'status' => 'pending',
);
}
foreach ( $this->rows as $row ) {
if ( $key === (string) ( $row['idempotency_key'] ?? '' ) ) {
return $row;
}
}
return null;
}

$job_id = (int) end( $args );
return $this->rows[ $job_id ] ?? null;
}
}

require_once dirname( __DIR__ ) . '/inc/Core/Database/BaseRepository.php';
require_once dirname( __DIR__ ) . '/inc/Core/Database/Jobs/Jobs.php';

use DataMachine\Core\Database\Jobs\Jobs;

$assertions = 0;
$assert = static function ( bool $condition, string $message ) use ( &$assertions ): void {
++$assertions;
if ( ! $condition ) {
throw new RuntimeException( "Assertion failed: {$message}" );
}
};
$new_jobs = static function ( string $mode ): array {
global $wpdb;
$wpdb = new Jobs_Idempotent_Insert_Test_Wpdb();
$wpdb->mode = $mode;
return array( new Jobs(), $wpdb );
};

list( $jobs, $wpdb ) = $new_jobs( 'success' );
$result = $jobs->create_or_get_job(
array(
'label' => str_repeat( 'x', 300 ),
'idempotency_key' => 'pipeline-batch:bounded-label',
)
);
$assert( is_array( $result ) && $result['created'], 'a valid idempotent job is created' );
$assert( 255 === mb_strlen( $wpdb->rows[1]['label'], 'UTF-8' ), 'label is bounded to its varchar(255) schema' );

list( $jobs, $wpdb ) = $new_jobs( 'duplicate_race' );
$result = $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:race' ) );
$assert( is_array( $result ) && 77 === $result['job_id'], 'a delayed duplicate-race winner becomes canonical' );
$assert( 1 === $wpdb->insert_calls, 'duplicate errors do not retry the insert' );
$assert( 3 === $wpdb->lookup_calls, 'duplicate winner visibility polling is bounded' );

list( $jobs, $wpdb ) = $new_jobs( 'contention_then_success' );
$result = $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:transient' ) );
$assert( is_array( $result ) && $result['created'], 'transient contention can recover' );
$assert( 3 === $wpdb->insert_calls, 'contention retries stop at the bounded successful attempt' );

$GLOBALS['job_insert_test_actions'] = array();
$GLOBALS['job_insert_test_cache'] = array();
list( $jobs, $wpdb ) = $new_jobs( 'contention' );
for ( $call = 0; $call < 5; ++$call ) {
$assert( false === $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:persistent' ) ), 'persistent contention remains visible to its caller' );
}
$logs = $GLOBALS['job_insert_test_actions']['datamachine_log'] ?? array();
$signals = $GLOBALS['job_insert_test_actions']['datamachine_idempotent_job_insert_failed'] ?? array();
$assert( 15 === $wpdb->insert_calls, 'persistent contention is bounded to three attempts per call' );
$assert( 1 === count( $logs ), 'identical persistent failures produce one log per throttle window' );
$assert( 5 === count( $signals ), 'every persistent failure remains observable to metrics listeners' );
$assert( 3 === $logs[0][2]['attempts'], 'the final diagnostic records retry exhaustion' );
$assert( str_contains( $logs[0][2]['db_error'], 'Deadlock found' ), 'the original insert error survives recovery lookups' );
$assert( ! isset( $logs[0][2]['idempotency_key'] ), 'diagnostics do not expose raw idempotency keys' );

$GLOBALS['job_insert_test_actions'] = array();
$GLOBALS['job_insert_test_cache'] = array();
list( $jobs, $wpdb ) = $new_jobs( 'non_retryable' );
$assert( false === $jobs->create_or_get_job( array( 'idempotency_key' => 'pipeline-batch:invalid' ) ), 'non-retryable database failures return false' );
$assert( 1 === $wpdb->insert_calls, 'non-retryable database failures are not amplified' );
$log = $GLOBALS['job_insert_test_actions']['datamachine_log'][0][2] ?? array();
$assert( 'non_retryable' === ( $log['error_type'] ?? '' ), 'non-retryable errors are classified safely' );

echo "jobs-idempotent-insert-recovery-smoke: {$assertions} assertions passed.\n";
}
Loading