Skip to content

fix: reconcile missing recurring flow schedules #2903

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
Jul 19, 2026
Merged

fix: reconcile missing recurring flow schedules #2903

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
6 changes: 4 additions & 2 deletions data-machine.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,6 +351,7 @@ function () {
new \DataMachine\Abilities\Flow\DuplicateFlowAbility();
new \DataMachine\Abilities\Flow\PauseFlowAbility();
new \DataMachine\Abilities\Flow\ResumeFlowAbility();
new \DataMachine\Abilities\Flow\ReconcileFlowSchedulesAbility();
new \DataMachine\Abilities\Flow\QueueAbility();
new \DataMachine\Abilities\Flow\WebhookTriggerAbility();
new \DataMachine\Abilities\FlowStep\GetFlowStepsAbility();
Expand Down Expand Up @@ -843,8 +844,9 @@ function datamachine_activate_for_site() {
// every deploy.
\DataMachine\Engine\AI\ComposableFileGenerator::regenerate_all();

// Re-schedule any flows with non-manual scheduling
datamachine_activate_scheduled_flows();
// Action Scheduler may not be initialized during activation. Mark this site
// for reconciliation on the next initialized scheduler request.
datamachine_mark_flow_schedule_reconciliation();

// Track DB schema version so deploy-time migrations auto-run.
update_option( 'datamachine_db_version', DATAMACHINE_VERSION, true );
Expand Down
84 changes: 84 additions & 0 deletions inc/Abilities/Flow/ReconcileFlowSchedulesAbility.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
<?php
/**
* Reconcile Flow Schedules Ability.
*
* @package DataMachine\Abilities\Flow
*/

namespace DataMachine\Abilities\Flow;

use DataMachine\Abilities\PermissionHelper;
use DataMachine\Api\Flows\FlowScheduleReconciler;

defined( 'ABSPATH' ) || exit;

class ReconcileFlowSchedulesAbility {

public function __construct() {
$this->registerAbility();
}

private function registerAbility(): void {
$register_callback = function () {
wp_register_ability(
'datamachine/reconcile-flow-schedules',
array(
'label' => __( 'Reconcile Flow Schedules', 'data-machine' ),
'description' => __( 'Audit recurring flow schedule coverage and optionally restore missing Action Scheduler actions.', 'data-machine' ),
'category' => 'datamachine-flow',
'input_schema' => array(
'type' => 'object',
'properties' => array(
'apply' => array(
'type' => 'boolean',
'default' => false,
'description' => __( 'Restore missing schedules. Defaults to dry-run.', 'data-machine' ),
),
'spread_hours' => array(
'type' => array( 'integer', 'null' ),
'minimum' => 1,
'maximum' => 24,
'description' => __( 'Explicit fleet distribution window in hours (1-24).', 'data-machine' ),
),
),
),
'output_schema' => array(
'type' => 'object',
'properties' => array(
'success' => array( 'type' => 'boolean' ),
'transient' => array( 'type' => 'boolean' ),
'code' => array( 'type' => 'string' ),
'applied' => array( 'type' => 'boolean' ),
'eligible' => array( 'type' => 'integer' ),
'covered' => array( 'type' => 'integer' ),
'missing' => array( 'type' => 'integer' ),
'blocked' => array( 'type' => 'integer' ),
'repaired' => array( 'type' => 'integer' ),
'failed' => array( 'type' => 'integer' ),
'remaining_missing' => array( 'type' => 'integer' ),
'invalid' => array( 'type' => 'integer' ),
'details' => array( 'type' => 'array' ),
'error' => array( 'type' => 'string' ),
),
),
'execute_callback' => array( $this, 'execute' ),
'permission_callback' => fn() => PermissionHelper::can_manage(),
'meta' => array( 'show_in_rest' => true ),
)
);
};

\DataMachine\Abilities\AbilityRegistration::on_abilities_api_init( $register_callback );
}

/**
* Execute schedule reconciliation.
*
* @param array $input Ability input.
* @return array Reconciliation report.
*/
public function execute( array $input ): array {
$spread_hours = isset( $input['spread_hours'] ) ? (int) $input['spread_hours'] : null;
return ( new FlowScheduleReconciler() )->reconcile( ! empty( $input['apply'] ), $spread_hours );
}
}
64 changes: 61 additions & 3 deletions inc/Abilities/SystemAbilities.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
use DataMachine\Core\Database\Flows\Flows;
use DataMachine\Core\Database\Pipelines\Pipelines;
use DataMachine\Core\PluginSettings;
use DataMachine\Api\Flows\FlowScheduleReconciler;
use DataMachine\Engine\Tasks\RecurringRejectionTracker;
use DataMachine\Engine\Tasks\RecurringScheduler;
use DataMachine\Engine\Tasks\TaskScheduler;
Expand Down Expand Up @@ -249,6 +250,21 @@ private function runSchedulerDiagnostics( array $options = array() ): array {
$complete = $this->getActionSchedulerGroupStats( 'complete' );
$due = $this->getActionSchedulerGroupStats( 'pending', true );
$daily_memory = $this->getActionSchedulerHookStats( 'datamachine_recurring_daily_memory_generation' );
$flow_coverage = array(
'success' => false,
'eligible' => 0,
'covered' => 0,
'remaining_missing' => 0,
'blocked' => 0,
'invalid' => 0,
);
if ( function_exists( 'as_get_scheduled_actions' ) && RecurringScheduler::isReady() ) {
$flow_coverage = ( new FlowScheduleReconciler() )->reconcile();
}
$missing_flow_schedules = (int) ( $flow_coverage['remaining_missing'] ?? 0 );
$invalid_flow_schedules = (int) ( $flow_coverage['invalid'] ?? 0 );
$blocked_flow_schedules = (int) ( $flow_coverage['blocked'] ?? 0 );
$coverage_failed = empty( $flow_coverage['success'] ) && 0 === $blocked_flow_schedules;

// Recurring bindings rejected by TaskScheduler::schedule() on every
// tick are a silent, permanent failure: the task never runs and the
Expand All @@ -262,13 +278,46 @@ private function runSchedulerDiagnostics( array $options = array() ): array {
$status = 'unavailable';
} elseif ( ! empty( $rejected_schedules ) ) {
$status = 'failing';
} elseif ( $blocked_flow_schedules > 0 || $coverage_failed ) {
$status = 'failing';
} elseif ( $missing_flow_schedules > 0 || $invalid_flow_schedules > 0 ) {
$status = 'failing';
} elseif ( $due['count'] > 0 || ( null !== $wp_cron_overdue && $wp_cron_overdue > $stale_threshold ) ) {
$status = 'stale';
}

$message = match ( $status ) {
'ok' => __( 'scheduler current', 'data-machine' ),
'failing' => sprintf(
'failing' => $blocked_flow_schedules > 0 ? sprintf(
/* translators: %d: number of flows blocked by in-progress Action Scheduler ownership. */
_n(
'%d recurring flow is blocked by in-progress Action Scheduler ownership',
'%d recurring flows are blocked by in-progress Action Scheduler ownership',
$blocked_flow_schedules,
'data-machine'
),
$blocked_flow_schedules
) : ( $missing_flow_schedules > 0 ? sprintf(
/* translators: %d: number of enabled recurring flows missing scheduler coverage. */
_n(
'%d enabled recurring flow is missing Action Scheduler coverage',
'%d enabled recurring flows are missing Action Scheduler coverage',
$missing_flow_schedules,
'data-machine'
),
$missing_flow_schedules
) : ( $invalid_flow_schedules > 0 ? sprintf(
/* translators: %d: number of malformed recurring flow definitions. */
_n(
'%d recurring flow has an invalid schedule definition',
'%d recurring flows have invalid schedule definitions',
$invalid_flow_schedules,
'data-machine'
),
$invalid_flow_schedules
) : ( $coverage_failed
? __( 'flow schedule coverage audit failed', 'data-machine' )
: sprintf(
/* translators: %d: number of recurring schedules rejected on every tick. */
_n(
'%d recurring schedule rejected on every tick and never running',
Expand All @@ -277,7 +326,7 @@ private function runSchedulerDiagnostics( array $options = array() ): array {
'data-machine'
),
count( $rejected_schedules )
),
) ) ) ),
'stale' => __( 'scheduler has overdue work; invoke WordPress cron, run wp datamachine drain, or run a specific task with wp datamachine system run <task_type> --wait', 'data-machine' ),
'unavailable' => __( 'Action Scheduler is unavailable', 'data-machine' ),
default => __( 'scheduler status unknown', 'data-machine' ),
Expand Down Expand Up @@ -323,8 +372,17 @@ private function runSchedulerDiagnostics( array $options = array() ): array {
'next_pending_gmt' => $daily_memory['pending']['oldest_gmt'],
'last_attempt_gmt' => $daily_memory['complete']['last_attempt_gmt'],
),
'flow_schedule_coverage' => $flow_coverage,
'recommendation' => match ( $status ) {
'failing' => __( 'One or more recurring schedules are rejected on every tick and never run. Inspect them with wp datamachine logs (filter error_code recurring_schedule_persistently_rejected) and fix the binding or its prerequisites (agent ownership, task registration, ability availability).', 'data-machine' ),
'failing' => $blocked_flow_schedules > 0
? __( 'Recover stale or uncertain in-progress jobs/actions first, then rerun wp datamachine flows reconcile-schedules. Reconciliation will not schedule around active ownership.', 'data-machine' )
: ( $missing_flow_schedules > 0
? __( 'Run wp datamachine flows reconcile-schedules to inspect missing coverage, then rerun with --apply. Large repairs automatically spread first runs across 24 hours.', 'data-machine' )
: ( $invalid_flow_schedules > 0
? __( 'Inspect invalid definitions with wp datamachine flows reconcile-schedules and correct each flow schedule before applying repairs.', 'data-machine' )
: ( $coverage_failed
? __( 'Run wp datamachine flows reconcile-schedules --format=json to inspect the coverage query failure before applying repairs.', 'data-machine' )
: __( 'One or more recurring schedules are rejected on every tick and never run. Inspect them with wp datamachine logs (filter error_code recurring_schedule_persistently_rejected) and fix the binding or its prerequisites (agent ownership, task registration, ability availability).', 'data-machine' ) ) ) ),
'stale' => __( 'For local or low-traffic installs, schedule an external wake-up such as wp cron event run --due-now. To remediate only daily memory for an affected agent, run wp datamachine system run daily_memory_generation --param=agent_slug=<slug> --wait; for all overdue work, run wp datamachine drain.', 'data-machine' ),
default => null,
},
Expand Down
Loading