* @license GPL-2.0-or-later */ defined( 'ABSPATH' ) || exit; /** * Class Studiou_WCMQ_Worker */ class Studiou_WCMQ_Worker { /** * Last WP_Error captured from wp_mail_failed, for the row in flight. * * @var string|null */ private $last_error = null; /** * Constructor. * * The wp_mail_failed listener is registered ONCE here, never per send. * Under Action Scheduler many send() calls run in one PHP process; a * per-send add_action leaks a closure per row, and an early `return` that * skips the paired remove_action leaves a stale WP_Error in the buffer to be * misattributed to the next row. Resetting the buffer before each send (in * send(), step 3) makes both failure modes structurally impossible. */ public function __construct() { add_action( Studiou_WCMQ_Queue::HOOK, array( $this, 'send' ), 10, 1 ); add_action( 'wp_mail_failed', array( $this, 'capture' ), 10, 1 ); } /** * Capture a mail error. * * @param WP_Error $error Error. */ public function capture( $error ) { if ( is_wp_error( $error ) ) { $this->last_error = $error->get_error_message(); } } /** * Send one queued mail. * * @param int $row_id Queue row id. */ public function send( $row_id ) { $row_id = (int) $row_id; if ( ! $row_id ) { return; } // The reaper runs before we claim OUR row, which looks unsafe: it hunts // pending rows with no scheduled action, and our row is pending and about // to run. It does not misfire, but only by a coincidence of Action // Scheduler internals — as_has_scheduled_action() matches STATUS_RUNNING // as well as STATUS_PENDING (packages/action-scheduler/functions.php:404), // and AS calls log_execution() (which sets RUNNING) before invoking this // hook (ActionScheduler_Abstract_QueueRunner.php:102). So our own row is // seen as running, not as an orphan. If AS ever changes that ordering, // this reaper call would duplicate the running row's action — revisit here. Studiou_WCMQ_Reaper::run(); // Atomic claim. Zero affected rows means someone else has it, or it is // already sent. This is the duplicate-send guard. if ( ! Studiou_WCMQ_DB::claim( $row_id ) ) { return; } $row = Studiou_WCMQ_DB::get_row( $row_id ); if ( ! $row ) { return; } $payload = json_decode( (string) $row->payload, true ); if ( ! is_array( $payload ) || JSON_ERROR_NONE !== json_last_error() || ! isset( $payload['to'], $payload['subject'], $payload['message'] ) ) { // Not retriable. Do not burn an attempt on it. $affected = Studiou_WCMQ_DB::update_row_if_state( $row_id, array( 'state' => Studiou_WCMQ_DB::STATE_FAILED, 'claimed_at' => null, 'last_error' => 'corrupt payload', ), Studiou_WCMQ_DB::STATE_SENDING ); Studiou_WCMQ_DB::log( 'error', 'Corrupt payload; row failed without sending.', $row_id ); // This is a terminal state. Without the note the order records nothing // at all: the customer got no email and the audit trail is silent. if ( 1 === (int) $affected ) { Studiou_WCMQ_Notifier::annotate( $row, false, 'corrupt payload' ); } Studiou_WCMQ_State::maybe_settle(); return; } $this->last_error = null; $sent = false; try { $sent = $this->dispatch( $payload ); } catch ( Throwable $e ) { $this->last_error = $e->getMessage(); $sent = false; } $error = $this->last_error; $this->last_error = null; // Completed send attempts, not crashes. Never bumped at claim, and folded // into the same UPDATE that commits the outcome so there is no window in // which the row has been attempted but not resolved. We hold the claim, so // nobody else is writing this column. $attempts = (int) $row->attempts + 1; if ( $sent ) { $this->on_success( $row, $row_id, $attempts ); } else { $this->on_failure( $row, $row_id, $attempts, $error ); } Studiou_WCMQ_State::maybe_settle(); } /** * Call wp_mail() with the stored payload, exactly once. * * Two hooks have to be neutralised for the duration: * * 1. `pre_wp_mail` — our own interceptor, or the mail would be re-queued * forever. Handled by the $sending flag. * * 2. `wp_mail` — WordPress applies this filter to $atts *before* it applies * `pre_wp_mail`, so the payload we snapshotted at enqueue has already been * through it. Calling wp_mail() again here would run every `wp_mail` filter * on the site a second time, against its own prior output: an unsubscribe * footer appended twice, a BCC-archive header duplicated. Queued mail would * differ from synchronous mail in ways nobody would connect to this plugin. * * Detaching the whole `wp_mail` hook and restoring it in `finally` is the only * way to suppress third-party callbacks we do not control. `wp_mail_from` and * friends need no such treatment — they are idempotent, and the resolved * sender is already baked into the stored headers. * * @param array $payload Decoded mail payload. * @return bool */ private function dispatch( array $payload ) { global $wp_filter; $headers = isset( $payload['headers'] ) ? $payload['headers'] : ''; $attachments = isset( $payload['attachments'] ) ? (array) $payload['attachments'] : array(); $saved_wp_mail = isset( $wp_filter['wp_mail'] ) ? $wp_filter['wp_mail'] : null; if ( null !== $saved_wp_mail ) { unset( $wp_filter['wp_mail'] ); } Studiou_WCMQ_Interceptor::$sending = true; try { return (bool) wp_mail( $payload['to'], $payload['subject'], $payload['message'], $headers, $attachments ); } finally { // Never leave either of these in place: under Action Scheduler many // sends share one PHP process, and a stuck flag or a missing hook // would corrupt every later mail in the batch. Studiou_WCMQ_Interceptor::$sending = false; if ( null !== $saved_wp_mail ) { $wp_filter['wp_mail'] = $saved_wp_mail; } } } /** * Commit the send, then annotate. * * Order matters. Once wp_mail() has returned true, nothing may throw back * into the send path: a fatal here would leave the row in `sending` for the * reaper to rescue and send a second time. * * @param object $row Row snapshot from before the send. * @param int $row_id Row id. * @param int $attempts Attempts including this one. */ private function on_success( $row, $row_id, $attempts ) { // State-guarded: if the row was deleted from the admin, or reclaimed by // the reaper, between our claim and now, this reports 0 rather than // silently "succeeding" and letting us delete attachments and write an // order note for a row that no longer exists. $affected = Studiou_WCMQ_DB::update_row_if_state( $row_id, array( 'state' => Studiou_WCMQ_DB::STATE_SENT, 'sent_at' => time(), 'claimed_at' => null, 'last_error' => null, 'attempts' => (int) $attempts, ), Studiou_WCMQ_DB::STATE_SENDING ); if ( 1 !== (int) $affected ) { // The mail went out but we could not record it. Either the DB errored // (row stays `sending`, the reaper will reclaim and resend — a // duplicate we cannot prevent from here) or the row vanished under us. // Either way: do not delete the attachment directory, so a resend // still carries its invoice, and write no note claiming success. Studiou_WCMQ_DB::log( 'error', 'Mail was sent but the row could not be committed to `sent` (it may have been deleted or reclaimed). It may be re-sent.', $row_id ); return; } Studiou_WCMQ_Queue::delete_attachment_dir( $row->attachment_dir ); Studiou_WCMQ_Notifier::annotate( $row, true, null ); Studiou_WCMQ_DB::log( 'info', sprintf( 'Sent %s to %s.', $row->context, $row->recipient ), $row_id ); } /** * Retry with backoff, or give up. * * @param object $row Row snapshot. * @param int $row_id Row id. * @param int $attempts Attempts after this one. * @param string|null $error Captured error. */ private function on_failure( $row, $row_id, $attempts, $error ) { $max_attempts = (int) Studiou_WCMQ_Settings::get( 'max_attempts' ); if ( $attempts >= $max_attempts ) { $affected = Studiou_WCMQ_DB::update_row_if_state( $row_id, array( 'state' => Studiou_WCMQ_DB::STATE_FAILED, 'claimed_at' => null, 'last_error' => $error, 'attempts' => (int) $attempts, ), Studiou_WCMQ_DB::STATE_SENDING ); // The attachment directory is deliberately KEPT: failed rows are // manually retriable from the admin, and a retry without its // attachments sends a broken email that looks fine. if ( 1 === (int) $affected ) { Studiou_WCMQ_Notifier::annotate( $row, false, $error ); } Studiou_WCMQ_DB::log( 'error', sprintf( 'Send failed permanently after %d attempts: %s', $attempts, (string) $error ), $row_id ); return; } $interval = Studiou_WCMQ_Settings::interval(); // `**`, not `^`. In PHP `^` is bitwise XOR and binds looser than `*`. $backoff = min( HOUR_IN_SECONDS, $interval * ( 2 ** $attempts ) ); list( $slot, $affected ) = Studiou_WCMQ_Queue::with_slot_lock( time() + $backoff, function ( $slot ) use ( $row_id, $error, $attempts ) { return Studiou_WCMQ_DB::update_row_if_state( $row_id, array( 'state' => Studiou_WCMQ_DB::STATE_PENDING, 'claimed_at' => null, 'scheduled_at' => $slot, 'last_error' => $error, 'attempts' => (int) $attempts, ), Studiou_WCMQ_DB::STATE_SENDING ); } ); if ( 1 !== (int) $affected ) { // The row did not move back to `pending`. Scheduling an action for it // would wake a worker that finds nothing to claim. The row is still // `sending`, so the reaper will reclaim it after claim_timeout. Studiou_WCMQ_DB::log( 'error', 'Could not return row to pending for retry; leaving it to the reaper.', $row_id ); return; } $action_id = as_schedule_single_action( $slot, Studiou_WCMQ_Queue::HOOK, array( (int) $row_id ), Studiou_WCMQ_Queue::GROUP ); if ( ! $action_id ) { // The row is pending with no action — an invariant violation. The // reaper repairs it, so this degrades to "recovered late", not "lost". Studiou_WCMQ_DB::log( 'error', 'Could not schedule retry; reaper will repair.', $row_id ); return; } Studiou_WCMQ_DB::log( 'warning', sprintf( 'Send failed (attempt %d/%d); retrying at %d. %s', $attempts, $max_attempts, $slot, (string) $error ), $row_id ); } }