qdr.app.woo.plugins/studiou-wc-mail-queue/includes/class-wcmq-reaper.php

<?php
/**
 * Recovers stranded queue rows.
 *
 * Enforces the invariant that makes the whole design durable:
 *
 *   Every row in `pending` has a scheduled Action Scheduler action.
 *
 * That invariant, not the enqueue rollback, is what prevents mail loss —
 * rollback cannot run after a PHP fatal.
 *
 * @package studiou-wc-mail-queue
 * @copyright 2026 QUADARAX
 * @author    Dalibor Votruba <dvotruba@quadarax.com>
 * @license   GPL-2.0-or-later
 */

defined( 'ABSPATH' ) || exit;

/**
 * Class Studiou_WCMQ_Reaper
 */
class Studiou_WCMQ_Reaper {

	const HOOK           = 'studiou_wcmq_reap';
	const INTERVAL       = 300;
	const ORPHAN_GRACE   = 300;
	const BATCH          = 200;
	const ADMIN_THROTTLE = 'studiou_wcmq_reap_throttle';

	/**
	 * Guard against running more than once per request.
	 *
	 * @var bool
	 */
	private static $ran_this_request = false;

	/**
	 * Register hooks.
	 */
	public static function init() {
		add_action( self::HOOK, array( __CLASS__, 'run' ) );
		add_action( 'admin_init', array( __CLASS__, 'maybe_run_from_admin' ) );
	}

	/**
	 * Schedule the recurring sweep, exactly once.
	 *
	 * Called from `init`, which runs on every request — so a bare
	 * as_schedule_recurring_action() here would register a fresh action each time.
	 */
	public static function schedule() {
		if ( ! function_exists( 'as_next_scheduled_action' ) || ! function_exists( 'as_schedule_recurring_action' ) ) {
			return;
		}
		if ( as_next_scheduled_action( self::HOOK, array(), Studiou_WCMQ_Queue::GROUP ) ) {
			return;
		}
		as_schedule_recurring_action(
			time() + self::INTERVAL,
			self::INTERVAL,
			self::HOOK,
			array(),
			Studiou_WCMQ_Queue::GROUP,
			true
		);
	}

	/**
	 * Heal from wp-admin, throttled.
	 *
	 * This is the trigger that still works when WP-Cron is broken — which is a
	 * plausible root cause of the wedge in the first place.
	 */
	public static function maybe_run_from_admin() {
		if ( get_transient( self::ADMIN_THROTTLE ) ) {
			return;
		}
		set_transient( self::ADMIN_THROTTLE, 1, 60 );
		self::run();
	}

	/**
	 * Sweep both classes of stranded row.
	 */
	public static function run() {
		if ( self::$ran_this_request ) {
			return;
		}
		self::$ran_this_request = true;

		if ( ! function_exists( 'as_has_scheduled_action' ) || ! function_exists( 'as_schedule_single_action' ) ) {
			return;
		}

		global $wpdb;
		$table   = Studiou_WCMQ_DB::queue_table();
		$now     = time();
		$timeout = (int) Studiou_WCMQ_Settings::get( 'claim_timeout' );

		// (a) `sending` rows whose worker died. Without this, one PHP fatal wedges
		// the queue forever: retention sweeps only sent/failed, the state machine
		// never reaches `disabled`, and the slot stays occupied.
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		$stale = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id, reclaims FROM {$table}
				  WHERE state = %s AND claimed_at IS NOT NULL AND claimed_at < %d
				  LIMIT %d",
				Studiou_WCMQ_DB::STATE_SENDING,
				$now - $timeout,
				self::BATCH
			)
		);
		if ( is_array( $stale ) ) {
			foreach ( $stale as $row ) {
				self::recover_crashed( (int) $row->id, (int) $row->reclaims );
			}
		}

		// (b) `pending` rows with no action — the §2.1 invariant violated. Arises
		// when a fatal lands between the INSERT and as_schedule_single_action(),
		// or when a retry's re-schedule failed. More likely than (a), because the
		// enqueue window is on the hot path of every bulk.
		// phpcs:ignore WordPress.DB.PreparedSQL.InterpolatedNotPrepared, WordPress.DB.DirectDatabaseQuery
		$orphans = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT id FROM {$table}
				  WHERE state = %s AND scheduled_at < %d
				  LIMIT %d",
				Studiou_WCMQ_DB::STATE_PENDING,
				$now - self::ORPHAN_GRACE,
				self::BATCH
			)
		);
		if ( is_array( $orphans ) ) {
			foreach ( $orphans as $row ) {
				// The cast is load-bearing. Action Scheduler matches args by
				// hashing json_encode($args); the action was scheduled with an int
				// so its hash is of "[123]", while $wpdb hands back the string
				// "123" whose hash is of "[\"123\"]". Without the cast this returns
				// false for a perfectly healthy row and we duplicate its action.
				$row_id = (int) $row->id;
				if ( as_has_scheduled_action( Studiou_WCMQ_Queue::HOOK, array( $row_id ), Studiou_WCMQ_Queue::GROUP ) ) {
					continue;
				}
				self::repair_orphan( $row_id );
			}
		}
	}

	/**
	 * A `pending` row lost its Action Scheduler action. Give it a new one.
	 *
	 * This is NOT a crashed worker and must not touch `reclaims`. The row was
	 * never claimed, never attempted, and nothing about it is suspect — the
	 * action went missing (a fatal between INSERT and schedule, a failed retry
	 * re-schedule, an admin pruning the AS tables). Charging it to the crash
	 * budget would eventually fail perfectly deliverable mail with
	 * "worker repeatedly crashed", which is both wrong and unactionable.
	 *
	 * There is no runaway risk: the re-slot moves scheduled_at into the future,
	 * so the next sweep will not see the row until the grace window elapses again.
	 *
	 * @param int $row_id Row id.
	 */
	private static function repair_orphan( $row_id ) {
		if ( ! self::reslot_and_schedule( $row_id, array(), Studiou_WCMQ_DB::STATE_PENDING ) ) {
			return;
		}
		Studiou_WCMQ_DB::log( 'warning', 'Pending row had no scheduled action; rescheduled.', $row_id );
	}

	/**
	 * A `sending` row whose worker died. Charge it to the crash budget.
	 *
	 * @param int $row_id   Row id.
	 * @param int $reclaims Reclaims so far.
	 */
	private static function recover_crashed( $row_id, $reclaims ) {
		$max_reclaims = (int) Studiou_WCMQ_Settings::get( 'max_reclaims' );
		$reclaims     = $reclaims + 1;

		// `reclaims`, not `attempts`. A worker killed by max_execution_time
		// strands whichever row happened to be running when the cumulative batch
		// budget expired — that is arbitrary, not a property of the row. Counting
		// crashes as send attempts permanently fails healthy mail that was never
		// once actually attempted.
		if ( $reclaims >= $max_reclaims ) {
			// Guarded on `sending`: if another sweep in a parallel process already
			// handled this row, we must not double-count or re-fail it.
			$affected = Studiou_WCMQ_DB::update_row_if_state(
				$row_id,
				array(
					'state'      => Studiou_WCMQ_DB::STATE_FAILED,
					'claimed_at' => null,
					'reclaims'   => $reclaims,
					'last_error' => 'worker repeatedly crashed while handling this mail',
				),
				Studiou_WCMQ_DB::STATE_SENDING
			);
			if ( 1 === (int) $affected ) {
				// Terminal state — the order must carry a note, or it records
				// nothing at all about a mail the customer never received.
				$row = Studiou_WCMQ_DB::get_row( $row_id );
				if ( $row ) {
					Studiou_WCMQ_Notifier::annotate( $row, false, 'worker repeatedly crashed while handling this mail' );
				}
				Studiou_WCMQ_DB::log( 'error', sprintf( 'Row failed after %d crash recoveries.', $reclaims ), $row_id );
			}
			return;
		}

		if ( ! self::reslot_and_schedule( $row_id, array( 'reclaims' => $reclaims ), Studiou_WCMQ_DB::STATE_SENDING ) ) {
			return;
		}

		Studiou_WCMQ_DB::log(
			'warning',
			sprintf( 'Recovered row from a dead worker; rescheduled. Crash recovery %d/%d.', $reclaims, $max_reclaims ),
			$row_id
		);
	}

	/**
	 * Return a row to `pending` on a fresh FUTURE slot and schedule it.
	 *
	 * Re-slotting is not optional. A row restored at its original scheduled_at
	 * now lies in the past, so Action Scheduler fires it on the next pass — and
	 * thirty recovered rows would all go out at once. That is the burst this
	 * plugin exists to prevent, delivered by its own recovery path.
	 *
	 * @param int    $row_id         Row id.
	 * @param array  $extra          Extra columns to set (e.g. reclaims).
	 * @param string $expected_state State the row must currently be in.
	 * @return bool True when the row is pending with a live action.
	 */
	private static function reslot_and_schedule( $row_id, array $extra, $expected_state ) {
		list( $slot, $affected ) = Studiou_WCMQ_Queue::with_slot_lock(
			time(),
			function ( $slot ) use ( $row_id, $extra, $expected_state ) {
				$data = array_merge(
					$extra,
					array(
						'state'        => Studiou_WCMQ_DB::STATE_PENDING,
						'claimed_at'   => null,
						'scheduled_at' => $slot,
					)
				);
				return Studiou_WCMQ_DB::update_row_if_state( $row_id, $data, $expected_state );
			}
		);

		// If the UPDATE changed no row — a DB error, or the row moved out from
		// under us — do NOT schedule. Scheduling anyway is how the reaper loops
		// forever: the row stays `sending` with a stale claimed_at and its
		// reclaims never persists, so the next sweep computes reclaims=0+1 again,
		// schedules another dead action, and never trips max_reclaims. Leave the
		// row for the next sweep, which will re-read its true reclaims from the DB.
		if ( 1 !== (int) $affected ) {
			Studiou_WCMQ_DB::log( 'error', 'Reaper could not move row to pending; not scheduling. Next sweep will retry.', $row_id );
			return false;
		}

		// Guarded exactly as enqueue and retry are. An unguarded call here would
		// re-create the orphan `pending` row the reaper exists to repair.
		$action_id = as_schedule_single_action( $slot, Studiou_WCMQ_Queue::HOOK, array( (int) $row_id ), Studiou_WCMQ_Queue::GROUP );
		if ( ! $action_id ) {
			Studiou_WCMQ_DB::log( 'error', 'Reaper could not reschedule; next sweep will retry.', $row_id );
			return false;
		}

		return true;
	}
}