Native: read MySQL claim candidates one status at a time, in index order #1065

Activity

1. rmcdaniel

   added

   kind:featureA public product capability or experience is requestedA public product capability or experience is requested

   priority:P0Immediate release or data-integrity riskImmediate release or data-integrity risk

   repo:serverOwned by the standalone server repositoryOwned by the standalone server repository

   agent:server

   on Nov 6, 2026
2. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0023-t14-Read-MySQL-s-claim-candidates-a-status-at-a-time-in-.patch, part 1/1.

   This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

   Complete sanitized patch: 7,434 bytes, SHA-256 a634db36345bef14a6615b6b4d8b7fb960f1bf545f8877b4ebfe3ee1843158d3. This part: 7,434 bytes, SHA-256 a634db36345bef14a6615b6b4d8b7fb960f1bf545f8877b4ebfe3ee1843158d3. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 90377282..7e5fc6eb 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -57,6 +57,8 @@ pub(super) trait Backend: Database {
        const WORKER_REGISTRATION_SQL: &str = super::store::WORKER_REGISTRATION_SQL;
        const TASK_CANDIDATES_SQL: &str = super::store::TASK_CANDIDATES_SQL;
        const TASK_CANDIDATE_BATCH_SQL: &str = super::store::TASK_CANDIDATE_BATCH_SQL;
   +    /// How many times a candidate query binds its poll's six values.
   +    const TASK_CANDIDATE_BRANCHES: usize = 1;
        const POLL_WAKE_TASKS_SQL: &str = super::cross_kind_poll::TASKS_SQL;
        const POLL_RECEIPT_CLEANUP_SQL: &str = super::store::POLL_RECEIPT_CLEANUP_SQL;
        const TIMER_TASK_SQL: &str;
   @@ -545,6 +547,26 @@ impl Backend for Postgres {
        }
    }
    
   +/// MySQL's `TASK_CANDIDATES_SQL` at most `$limit` candidates. One range over
   +/// both claimable statuses comes back out of claim order, so MySQL read every
   +/// claimable task in the queue, joined each to its run and sorted them all to
   +/// keep the first few. Each branch here is the shared filter at one status,
   +/// which the dispatch order index returns in claim order, so each branch stops
   +/// at `$limit`. Rows of the other task kind ahead in that order are read and
   +/// skipped. Each branch binds the poll's values again.
   +macro_rules! mysql_task_candidates_sql {
   +    ($limit:literal) => {
   +        concat!(
   +            "SELECT * FROM ((SELECT STRAIGHT_JOIN t.*,r.workflow_type FROM workflow_tasks t FORCE INDEX (workflow_tasks_dispatch_order_index) JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.status='ready' AND t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT ",
   +            $limit,
   +            ") UNION ALL (SELECT STRAIGHT_JOIN t.*,r.workflow_type FROM workflow_tasks t FORCE INDEX (workflow_tasks_dispatch_order_index) JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.status='leased' AND t.namespace=$7 AND t.queue=$8 AND t.task_type=$9 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$10)) AND (t.available_at IS NULL OR t.available_at<=$11) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$12) ORDER BY t.priority,t.available_at,t.id LIMIT ",
   +            $limit,
   +            ")) AS candidates ORDER BY priority,available_at,id LIMIT ",
   +            $limit
   +        )
   +    };
   +}
   +
    impl Backend for MySql {
        type EncodedInteger = Option<i64>;
        // MariaDB's Laravel driver may store the owner as its uuid type.
   @@ -644,6 +666,9 @@ impl Backend for MySql {
            "CAST(SUBSTRING(c.payload,1,30) AS BINARY)='dw-native-external-payload-v1:'",
            "CAST(SUBSTRING(c.payload,1,23) AS BINARY)='dw-external-payload:v1:'"
        );
   +    const TASK_CANDIDATES_SQL: &str = mysql_task_candidates_sql!(100);
   +    const TASK_CANDIDATE_BATCH_SQL: &str = mysql_task_candidates_sql!(10);
   +    const TASK_CANDIDATE_BRANCHES: usize = 2;
        // First use the preserved run/sequence index to materialize only IDs and
        // byte counts for this bounded page. Sorting JSON payloads in the window
        // exhausts MySQL's default sort buffer before the size rejection can run.
   @@ -792,6 +817,16 @@ mod timestamp_tests {
            );
        }
    
   +    #[test]
   +    fn mysql_candidate_queries_bind_the_poll_once_per_branch() {
   +        for sql in [MySql::TASK_CANDIDATES_SQL, MySql::TASK_CANDIDATE_BATCH_SQL] {
   +            assert_eq!(
   +                MySql::statement(sql).matches('?').count(),
   +                6 * MySql::TASK_CANDIDATE_BRANCHES
   +            );
   +        }
   +    }
   +
        #[test]
        #[should_panic(expected = "ordered unique positions")]
        fn mysql_statement_refuses_out_of_order_positions_after_a_json_path() {
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index a55c48cb..3ccf5ff3 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -2119,18 +2119,35 @@ where
            {
                return Ok(true);
            }
   -        let at = now();
   -        Ok(Self::query(DB::TASK_CANDIDATE_BATCH_SQL)
   -            .bind(self.namespace.as_ref())
   -            .bind(queue)
   -            .bind(kind)
   -            .bind(DB::bind_time(at))
   -            .bind(DB::bind_time(at))
   -            .bind(build.as_deref())
   +        Ok(self
   +            .task_candidates(DB::TASK_CANDIDATE_BATCH_SQL, queue, kind, build.as_deref(), now())
                .fetch_optional(&self.pool)
                .await?
                .is_some())
        }
   +    /// `sql`, a candidate query, bound to a poll as of `at`: the namespace,
   +    /// queue, kind, lease expiry and availability times, and build, once per
   +    /// `Backend::TASK_CANDIDATE_BRANCHES`.
   +    fn task_candidates(
   +        &self,
   +        sql: &'static str,
   +        queue: &str,
   +        kind: &str,
   +        build: Option<&str>,
   +        at: DateTime<Utc>,
   +    ) -> sqlx::query::Query<'static, DB, DB::Arguments> {
   +        let mut query = Self::query(sql);
   +        for _ in 0..DB::TASK_CANDIDATE_BRANCHES {
   +            query = query
   +                .bind(self.namespace.as_ref())
   +                .bind(queue)
   +                .bind(kind)
   +                .bind(DB::bind_time(at))
   +                .bind(DB::bind_time(at))
   +                .bind(build);
   +        }
   +        query
   +    }
    
        /// Published PHP's poll order (`claimReadyTask`): its bridge returns the
        /// candidates of supported workflow types a batch at a time, and the
   @@ -2327,13 +2344,8 @@ where
            // the whole candidate list from the same instant, and the candidates
            // already tried are not tried again.
            let at = now();
   -        let first = Self::query(DB::TASK_CANDIDATE_BATCH_SQL)
   -            .bind(self.namespace.as_ref())
   -            .bind(queue)
   -            .bind(kind)
   -            .bind(DB::bind_time(at))
   -            .bind(DB::bind_time(at))
   -            .bind(build.as_deref())
   +        let first = self
   +            .task_candidates(DB::TASK_CANDIDATE_BATCH_SQL, queue, kind, build.as_deref(), at)
                .fetch_all(&mut **tx)
                .await?;
            // A short batch is every candidate there is.
   @@ -2351,13 +2363,8 @@ where
                    if !std::mem::take(&mut unread) {
                        break;
                    }
   -                let all = Self::query(DB::TASK_CANDIDATES_SQL)
   -                    .bind(self.namespace.as_ref())
   -                    .bind(queue)
   -                    .bind(kind)
   -                    .bind(DB::bind_time(at))
   -                    .bind(DB::bind_time(at))
   -                    .bind(build.as_deref())
   +                let all = self
   +                    .task_candidates(DB::TASK_CANDIDATES_SQL, queue, kind, build.as_deref(), at)
                        .fetch_all(&mut **tx)
                        .await?;
                    let mut rest = Vec::new();
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized rebased reference for report topic 14: 0023-t14-Read-MySQL-s-claim-candidates-a-status-at-a-time-in-.patch.

   The supplied revision targets Server b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. It is cumulative external reference material, not independently qualified against today's maintained source. Mail headers and commit-message/application instructions were removed; the unified diff is retained. The reporter now claims every rebased commit builds and the tip passes SQLite/MySQL/PostgreSQL tests. Those are attributed claims, not a new maintainer test result. Existing acceptance criteria, security findings and completed implementation receipts remain authoritative. Earlier attached patches are retained as historical references.

   Sanitized patch: 7,434 bytes, SHA-256 e123fc54f55b627e562a5b8afd61e91477443e756084256ad10298d81cd3799b. Original-base sanitized identity: a634db36345bef14a6615b6b4d8b7fb960f1bf545f8877b4ebfe3ee1843158d3. No patch was applied or executed during this update.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 90377282..7e5fc6eb 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -57,6 +57,8 @@ pub(super) trait Backend: Database {
        const WORKER_REGISTRATION_SQL: &str = super::store::WORKER_REGISTRATION_SQL;
        const TASK_CANDIDATES_SQL: &str = super::store::TASK_CANDIDATES_SQL;
        const TASK_CANDIDATE_BATCH_SQL: &str = super::store::TASK_CANDIDATE_BATCH_SQL;
   +    /// How many times a candidate query binds its poll's six values.
   +    const TASK_CANDIDATE_BRANCHES: usize = 1;
        const POLL_WAKE_TASKS_SQL: &str = super::cross_kind_poll::TASKS_SQL;
        const POLL_RECEIPT_CLEANUP_SQL: &str = super::store::POLL_RECEIPT_CLEANUP_SQL;
        const TIMER_TASK_SQL: &str;
   @@ -545,6 +547,26 @@ impl Backend for Postgres {
        }
    }
    
   +/// MySQL's `TASK_CANDIDATES_SQL` at most `$limit` candidates. One range over
   +/// both claimable statuses comes back out of claim order, so MySQL read every
   +/// claimable task in the queue, joined each to its run and sorted them all to
   +/// keep the first few. Each branch here is the shared filter at one status,
   +/// which the dispatch order index returns in claim order, so each branch stops
   +/// at `$limit`. Rows of the other task kind ahead in that order are read and
   +/// skipped. Each branch binds the poll's values again.
   +macro_rules! mysql_task_candidates_sql {
   +    ($limit:literal) => {
   +        concat!(
   +            "SELECT * FROM ((SELECT STRAIGHT_JOIN t.*,r.workflow_type FROM workflow_tasks t FORCE INDEX (workflow_tasks_dispatch_order_index) JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.status='ready' AND t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT ",
   +            $limit,
   +            ") UNION ALL (SELECT STRAIGHT_JOIN t.*,r.workflow_type FROM workflow_tasks t FORCE INDEX (workflow_tasks_dispatch_order_index) JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.status='leased' AND t.namespace=$7 AND t.queue=$8 AND t.task_type=$9 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$10)) AND (t.available_at IS NULL OR t.available_at<=$11) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$12) ORDER BY t.priority,t.available_at,t.id LIMIT ",
   +            $limit,
   +            ")) AS candidates ORDER BY priority,available_at,id LIMIT ",
   +            $limit
   +        )
   +    };
   +}
   +
    impl Backend for MySql {
        type EncodedInteger = Option<i64>;
        // MariaDB's Laravel driver may store the owner as its uuid type.
   @@ -644,6 +666,9 @@ impl Backend for MySql {
            "CAST(SUBSTRING(c.payload,1,30) AS BINARY)='dw-native-external-payload-v1:'",
            "CAST(SUBSTRING(c.payload,1,23) AS BINARY)='dw-external-payload:v1:'"
        );
   +    const TASK_CANDIDATES_SQL: &str = mysql_task_candidates_sql!(100);
   +    const TASK_CANDIDATE_BATCH_SQL: &str = mysql_task_candidates_sql!(10);
   +    const TASK_CANDIDATE_BRANCHES: usize = 2;
        // First use the preserved run/sequence index to materialize only IDs and
        // byte counts for this bounded page. Sorting JSON payloads in the window
        // exhausts MySQL's default sort buffer before the size rejection can run.
   @@ -792,6 +817,16 @@ mod timestamp_tests {
            );
        }
    
   +    #[test]
   +    fn mysql_candidate_queries_bind_the_poll_once_per_branch() {
   +        for sql in [MySql::TASK_CANDIDATES_SQL, MySql::TASK_CANDIDATE_BATCH_SQL] {
   +            assert_eq!(
   +                MySql::statement(sql).matches('?').count(),
   +                6 * MySql::TASK_CANDIDATE_BRANCHES
   +            );
   +        }
   +    }
   +
        #[test]
        #[should_panic(expected = "ordered unique positions")]
        fn mysql_statement_refuses_out_of_order_positions_after_a_json_path() {
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 532058bf..495695d3 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -2127,18 +2127,35 @@ where
            {
                return Ok(true);
            }
   -        let at = now();
   -        Ok(Self::query(DB::TASK_CANDIDATE_BATCH_SQL)
   -            .bind(self.namespace.as_ref())
   -            .bind(queue)
   -            .bind(kind)
   -            .bind(DB::bind_time(at))
   -            .bind(DB::bind_time(at))
   -            .bind(build.as_deref())
   +        Ok(self
   +            .task_candidates(DB::TASK_CANDIDATE_BATCH_SQL, queue, kind, build.as_deref(), now())
                .fetch_optional(&self.pool)
                .await?
                .is_some())
        }
   +    /// `sql`, a candidate query, bound to a poll as of `at`: the namespace,
   +    /// queue, kind, lease expiry and availability times, and build, once per
   +    /// `Backend::TASK_CANDIDATE_BRANCHES`.
   +    fn task_candidates(
   +        &self,
   +        sql: &'static str,
   +        queue: &str,
   +        kind: &str,
   +        build: Option<&str>,
   +        at: DateTime<Utc>,
   +    ) -> sqlx::query::Query<'static, DB, DB::Arguments> {
   +        let mut query = Self::query(sql);
   +        for _ in 0..DB::TASK_CANDIDATE_BRANCHES {
   +            query = query
   +                .bind(self.namespace.as_ref())
   +                .bind(queue)
   +                .bind(kind)
   +                .bind(DB::bind_time(at))
   +                .bind(DB::bind_time(at))
   +                .bind(build);
   +        }
   +        query
   +    }
    
        /// Published PHP's poll order (`claimReadyTask`): its bridge returns the
        /// candidates of supported workflow types a batch at a time, and the
   @@ -2335,13 +2352,8 @@ where
            // the whole candidate list from the same instant, and the candidates
            // already tried are not tried again.
            let at = now();
   -        let first = Self::query(DB::TASK_CANDIDATE_BATCH_SQL)
   -            .bind(self.namespace.as_ref())
   -            .bind(queue)
   -            .bind(kind)
   -            .bind(DB::bind_time(at))
   -            .bind(DB::bind_time(at))
   -            .bind(build.as_deref())
   +        let first = self
   +            .task_candidates(DB::TASK_CANDIDATE_BATCH_SQL, queue, kind, build.as_deref(), at)
                .fetch_all(&mut **tx)
                .await?;
            // A short batch is every candidate there is.
   @@ -2359,13 +2371,8 @@ where
                    if !std::mem::take(&mut unread) {
                        break;
                    }
   -                let all = Self::query(DB::TASK_CANDIDATES_SQL)
   -                    .bind(self.namespace.as_ref())
   -                    .bind(queue)
   -                    .bind(kind)
   -                    .bind(DB::bind_time(at))
   -                    .bind(DB::bind_time(at))
   -                    .bind(build.as_deref())
   +                let all = self
   +                    .task_candidates(DB::TASK_CANDIDATES_SQL, queue, kind, build.as_deref(), at)
                        .fetch_all(&mut **tx)
                        .await?;
                    let mut rest = Vec::new();
4. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Current implementation: draft PR #1085, exact 72d6983f766640cf09c82cc9cd924af24ab49408 / tree 6a3d21971f92eaf31886a60909e3d57c336ee397, is based on accepted main 4fb4cffc. Each MySQL status range retains the original filters and 100-row limit through the existing dispatch index. The bounded merge retains the original order and page limit. No reference patch was applied.

   Server owns the sole current source CI train, real MySQL/MariaDB candidate parity and EXPLAIN controls, SQLite/PostgreSQL preservation, normal integration/main acceptance and topic cleanup. Hosted native compilation has passed; full backend/PHP suites and actual plans remain pending. #1069 stays local until this train is terminal. Release/published acceptance remains open; no benchmark speedup or installed result is claimed.
5. rmcdaniel commented on Nov 8, 2026

   

   rmcdaniel

   on Nov 8, 2026

   MemberAuthor

   More actions

   Published-artifact reconciliation: delivered in Native Server3.3.0.

   Implementation #1085 is an ancestor of immutable release source 2a15179f29801eec9f2a2d4e6a9c5f93233f104e. Its accepted source checks are62success/1intended skip. The integrated #1098 source head 5f937ec8f385c71dc781875f2f38026b63277faa has a byte-identical rust/ tree to this release. I independently read its successful SQLite, MySQL8.0, MariaDB10.11 and PostgreSQL17 native-test job commands: they run the registered all-targets/ignored qualification suites (only the unrelated MySQL bootstrap child is skipped). This supports the specific source controls below; it is not a new benchmark run.

   MySQL/MariaDB candidate equivalence and EXPLAIN controls require bounded per-status index order and prevent a queue-wide filesort. The later first-batch change keeps these checks for both limits.

   The 3.3.0 release is public at index sha256:cb4b11bd6235a551cf51e7dc6342504fbbcef19959618b376fb5648ca00fc388, with provenance binding M. Published SQLite corpus acceptance passed, as did four-cell MySQL Compose acceptance, the no-input default, and the release’s SQLite/PostgreSQL/MySQL startup/identity/stop smokes. Both registries’3.3/3/latest aliases resolve to D.

   The issue’s functional/source acceptance and release follow-through are met. Closing as completed. No isolated speedup, reporter benchmark reproduction, all-backend installed corpus or installed EXPLAIN rerun is claimed. Earlier rejected/combined subjects remain attributed. Remaining P0 work, including #1099, is excluded from this3.3.0 cut. Schema11 and rollback constraints remain in the tagged release notes.
6. rmcdaniel

   closed this as completedon Nov 8, 2026