Force stalled queues to resume

cloudinary-image-management-and-manipulation-in-the-cloud-cdn/php/sync/class-push-sync.php

@@ -40,6 +40,13 @@ class Push_Sync {
 	 */
 	private $sync_types;
 
+	/**
+	 * Holds the ID of the last attachment synced.
+	 *
+	 * @var int
+	 */
+	protected $post_id;
+
 	/**
 	 * Push_Sync constructor.
 	 *
@@ -200,19 +207,33 @@ public function rest_push_attachments( \WP_REST_Request $request ) {
 		if ( ! $this->plugin->components['sync']->managers['queue']->is_running() ) { // Check it wasn't stopped.
 			return $this->rest_get_queue_status();
 		}
-		$post_id = $this->plugin->components['sync']->managers['queue']->get_post();
+
+		$this->post_id = $this->plugin->components['sync']->managers['queue']->get_post();
+
 		// No post, end of queue.
-		if ( empty( $post_id ) ) {
+		if ( empty( $this->post_id ) ) {
 			$this->plugin->components['sync']->managers['queue']->stop_queue();
 
 			return $this->rest_get_queue_status();
 		}
 
+		add_action( 'shutdown', array( $this, 'resume_queue' ) );
+
+		return $this->rest_get_queue_status();
+	}
+
+	/**
+	 * Resume the bulk sync.
+	 *
+	 * @return bool|\WP_REST_Response
+	 */
+	public function resume_queue() {
 		// Check if there is a Cloudinary ID in case this was synced on-demand before being processed by the queue.
 		add_filter( 'cloudinary_on_demand_sync_enabled', '__return_false' ); // Disable the on-demand sync since we want the status.
 		add_filter( 'cloudinary_id', '__return_false' ); // Disable the on-demand sync since we want the status.
-		if ( false === $this->plugin->components['media']->cloudinary_id( $post_id ) ) {
-			$stat = $this->push_attachments( array( $post_id ) );
+
+		if ( false === $this->plugin->components['media']->cloudinary_id( $this->post_id ) ) {
+			$stat = $this->push_attachments( array( $this->post_id ) );
 			if ( ! empty( $stat['processed'] ) ) {
 				$result = 'done';
 			} else {
@@ -226,8 +247,7 @@ public function rest_push_attachments( \WP_REST_Request $request ) {
 			$result = 'done';
 		}
 
-		return $this->call_thread( $post_id, $result );
-
+		return $this->call_thread( $this->post_id, $result );
 	}
 
 	/**
@@ -246,7 +266,8 @@ private function call_thread( $last_id = null, $last_result = null ) {
 			$params['last_id']     = $last_id;
 			$params['last_result'] = $last_result;
 		}
-		// Setup background call to cintinue the queue.
+
+		// Setup background call to continue the queue.
 		$this->plugin->components['api']->background_request( 'process', $params );
 
 		return $this->rest_get_queue_status();

cloudinary-image-management-and-manipulation-in-the-cloud-cdn/php/sync/class-upload-queue.php

@@ -39,13 +39,41 @@ class Upload_Queue {
 	 */
 	private $running = null;
 
+	/**
+	 * The cron frequency to ensure that the queue is progressing.
+	 *
+	 * @var int
+	 */
+	protected $cron_frequency;
+
+	/**
+	 * The cron offset since the last update.
+	 *
+	 * @var int
+	 */
+	protected $cron_start_offset;
+
 	/**
 	 * Upload_Queue constructor.
 	 *
 	 * @param \Cloudinary\Plugin $plugin The plugin.
 	 */
 	public function __construct( \Cloudinary\Plugin $plugin ) {
 		$this->plugin = $plugin;
+
+		$this->cron_frequency    = apply_filters( 'cloudinary_cron_frequency', 10 * MINUTE_IN_SECONDS );
+		$this->cron_start_offset = apply_filters( 'cloudinary_cron_start_offset', MINUTE_IN_SECONDS );
+
+		$this->load_hooks();
+	}
+
+	/**
+	 * Load the Upload Queue hooks.
+	 *
+	 * @return void
+	 */
+	public function load_hooks() {
+		add_action( 'cloudinary_resume_queue', array( $this, 'maybe_resume_queue' ) );
 	}
 
 	/**
@@ -85,6 +113,7 @@ public function get_post() {
 			if ( ! empty( $queue['pending'] ) ) {
 				$id                    = array_shift( $queue['pending'] );
 				$queue['processing'][] = $id;
+				$queue['last_update']  = current_time( 'timestamp' );
 				$this->set_queue( $queue );
 			}
 		}
@@ -209,7 +238,6 @@ public function build_queue() {
 		// Transform attachments.
 		$return = array(
 			'pending'    => array(),
-			'large'      => array(),
 			'done'       => array(),
 			'processing' => array(),
 			'error'      => array(),
@@ -233,17 +261,40 @@ public function stop_queue() {
 			$this->set_queue( $queue );
 		}
 		$this->running = false;
-	}
 
+		wp_clear_scheduled_hook( 'cloudinary_resume_queue' );
+	}
 
 	/**
-	 * Startthe queue by setting the started flag.
+	 * Start the queue by setting the started flag.
+	 *
+	 * @return void
 	 */
 	public function start_queue() {
 		$queue            = $this->get_queue();
-		$queue['started'] = current_time( 'timestamp' );
+		$queue['started'] = $queue['last_update'] = current_time( 'timestamp' );
 		$this->set_queue( $queue );
 		$this->running = true;
+
+		if ( ! wp_next_scheduled( 'cloudinary_resume_queue' ) ) {
+			wp_schedule_single_event( time() + $this->cron_frequency, 'cloudinary_resume_queue' );
+		}
 	}
 
+	/**
+	 * Maybe resume the queue.
+	 * This is a fallback mechanism to resume the queue when it stops unexpectedly.
+	 *
+	 * @return void
+	 */
+	public function maybe_resume_queue() {
+		$now   = current_time( 'timestamp' );
+		$queue = $this->get_queue();
+
+		if ( $now - $queue['last_update'] > $this->cron_start_offset && $this->is_running() ) {
+			$this->plugin->components['api']->background_request( 'process', array() );
+		}
+
+		wp_schedule_single_event( $now + $this->cron_frequency, 'cloudinary_resume_queue' );
+	}
 }