Store MySQL JSON documents as exact text (#1016)

rust/README.md

@@ -355,6 +355,19 @@ completing worker's contract once, with its first accepted workflow task
 completion. Until then its updates return HTTP 409 `contract_pending`. Stop
 the native nodes, back up the database and run `schema-upgrade` before
 starting this binary on an earlier schema.
+MySQL schema 13 (Server #1016) stores every JSON document column as exact
+`utf8mb4_bin` text with an enforced `json_valid` check, as MariaDB already
+does, so doubles keep every digit. MySQL's `json` type parsed doubles itself
+and could store a value 1 ULP off. Only MySQL changes: MariaDB and PostgreSQL stay
+at schema 12 and SQLite at 11. The upgrade rebuilds the 36 tables that hold
+JSON columns, one atomic `ALTER TABLE` each, and an interrupted upgrade resumes
+table by table. On MySQL: stop every native node, back up the database, run
+`schema-upgrade` once, then start the new nodes. The service is down for the
+whole upgrade, and the time grows with the data: measured on MySQL 8.0.46 with
+2 CPUs and the default 128 MiB buffer pool, 1,000,000 history rows (2.65 GB)
+took 115 s, about 2 minutes per million history rows; plan for the other JSON
+tables too. Documents already stored keep MySQL's rendering, including any
+rounded doubles. Rolling back means restoring the backup.
 
 Implemented paths cover health/schema readiness, limited capability discovery,
 workflow start/describe/paginated history, worker registration/heartbeat/removal,

rust/migrations/mariadb/0012_checksum.txt

@@ -0,0 +1 @@
+477621f4b2083ef963b6db6ed530e7eb5d72704f32f250f0640bb99a770f420f78f13277fcd2cc5cb23a5928965cd9e2

rust/migrations/mysql/0012_checksum.txt

@@ -0,0 +1 @@
+f5430aad3bb99703601105763a97089326f4e842792eb98b15518d069aedf4bb8582add033aeae66e68c14f49bcc6a64

rust/src/runtime/backend.rs

@@ -793,8 +793,9 @@ impl Backend for MySql {
         let mut tx = pool.begin().await?;
         // InnoDB row locking releases on commit/rollback/process loss. This
         // first slice serializes transitions on its single ownership row.
-        sqlx::query("SELECT engine FROM dw_server_schema WHERE engine='rust-development' AND version=? FOR UPDATE")
-            .bind(mysql::VERSION).fetch_one(&mut *tx).await?;
+        // MariaDB stays at 12 while MySQL moved to 13 (server#1016).
+        sqlx::query("SELECT engine FROM dw_server_schema WHERE engine='rust-development' AND version=IF(VERSION() LIKE '%MariaDB%',?,?) FOR UPDATE")
+            .bind(mysql::MARIADB_VERSION).bind(mysql::MYSQL_VERSION).fetch_one(&mut *tx).await?;
         Ok(tx)
     }
     async fn history_ready(pool: &Pool<Self>) -> bool {

rust/src/runtime/schema_upgrade.rs

@@ -30,7 +30,7 @@ pub(super) const DEVELOPMENT_ENGINE: &str = "rust-development";
 pub(super) const UPGRADING_ENGINE: &str = "rust-upgrading";
 /// Earlier native schema versions this binary upgrades in place. Release
 /// candidates `3.0.0-rc.1` and `3.0.0-rc.2` shipped version 7.
-pub(super) const UPGRADABLE: &[i64] = &[7, 8, 10, 11];
+pub(super) const UPGRADABLE: &[i64] = &[7, 8, 10, 11, 12];
 
 /// The refusal for a native marker this binary does not open as it is.
 pub(super) fn marker_refusal(engine: &str, version: i64, current: i64) -> RuntimeError {
@@ -131,8 +131,8 @@ pub async fn upgrade_postgres(options: PgConnectOptions) -> Result<Value> {
 /// Upgrade a native MySQL or MariaDB database under the bootstrap named
 /// lock. An interrupted upgrade resumes when the command is run again.
 pub async fn upgrade_mysql(options: MySqlConnectOptions) -> Result<Value> {
-    let (backend, from) = super::mysql::upgrade(options).await?;
-    Ok(report(backend, from, super::mysql::VERSION))
+    let (backend, from, to) = super::mysql::upgrade(options).await?;
+    Ok(report(backend, from, to))
 }
 
 #[cfg(test)]

rust/src/runtime/upgrade/takeover/mysql_journal.rs

@@ -459,6 +459,7 @@ async fn finish(connection: &mut MySqlConnection, run: &Run<'_>) -> Result<()> {
     mysql::ensure_ready_origin(connection, run.flavor).await?;
     mysql::ensure_scoped_dispatch_index(connection, run.flavor).await?;
     mysql::ensure_run_command_contracts(connection, run.flavor).await?;
+    mysql::ensure_json_text(connection, run.flavor).await?;
     connection.execute(LEDGER_SQL).await?;
     let migration = run.flavor.migration();
     let mut tx = connection.begin().await?;
@@ -469,7 +470,7 @@ async fn finish(connection: &mut MySqlConnection, run: &Run<'_>) -> Result<()> {
         .execute(&mut *tx)
         .await?;
     let marked = sqlx::query("UPDATE dw_server_schema SET engine='rust-development',version=?,bootstrap_checksum=? WHERE engine=? AND version=?")
-        .bind(mysql::VERSION)
+        .bind(run.flavor.version())
         .bind(run.flavor.checksum())
         .bind(format!("{TAKEOVER_ENGINE}{}", run.layout))
         .bind(LAST)

rust/tests/execution.rs

@@ -212,6 +212,8 @@ mod command_batches;
 mod child_open_parent_task;
 #[path = "execution/update_routing.rs"]
 mod update_routing;
+#[path = "execution/json_doubles.rs"]
+mod json_doubles;
 #[path = "execution/update_reads.rs"]
 mod update_reads;
 #[path = "execution/command_payloads.rs"]

rust/tests/execution/json_doubles.rs

@@ -0,0 +1,74 @@
+use super::*;
+
+// server#1016: JSON doubles come back bit-exact on every backend, MySQL
+// included, through the run's memo, its double search attributes (describe,
+// list and WorkflowStarted) and its activity heartbeat details. MySQL's json
+// type used to store its own binary double, one ULP off for some values;
+// schema 13 stores each document as exact utf8mb4_bin text.
+const DOUBLES: [&str; 6] = ["1.4820142857142857", "124435.71428571429", "33.333333333333336",
+    "0.7071067811865476", "1.0000000000000002", "2.2250738585072014e-308"];
+
+fn exact(found: &Value, key: &str, text: &str, surface: &str) {
+    let wanted: f64 = text.parse().unwrap();
+    let actual = found[key].as_f64().unwrap_or_else(|| panic!("{surface} {key} missing: {found}"));
+    assert_eq!(actual.to_bits(), wanted.to_bits(), "{surface} {key}: {actual} is not {text}");
+}
+
+#[tokio::test]
+async fn doubles_come_back_bit_exact_through_history_memo_search_attributes_and_heartbeats() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    if matches!(database, TestDatabase::MySql { .. }) {
+        let types = text_rows(&database, "", "", "SELECT COLUMN_NAME, DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='workflow_history_events' AND COLUMN_NAME='payload'").await;
+        assert_eq!(types, [("payload".to_owned(), "longtext".to_owned())]);
+    }
+    let mut memo = serde_json::Map::new();
+    let mut attributes = serde_json::Map::new();
+    for (index, text) in DOUBLES.iter().enumerate() {
+        let value: f64 = text.parse().unwrap();
+        memo.insert(format!("d{index}"), json!(value));
+        let name = format!("Dbl{index}");
+        let (status, body) = request(&app, "POST", "/api/search-attributes", json!({"name":name,"type":"double"})).await;
+        assert_eq!(status, StatusCode::CREATED, "{body}");
+        attributes.insert(name, json!(value));
+    }
+    let (status, body) = request(&app, "POST", "/api/worker/register", json!({"worker_id":"doubles-worker","task_queue":"test",
+        "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":["measure"]})).await;
+    assert_eq!(status, StatusCode::CREATED, "{body}");
+    let (status, started) = request(&app, "POST", "/api/workflows", json!({"workflow_id":"doubles","workflow_type":"echo",
+        "task_queue":"test","input":envelope(Payload::Array(vec![])),"memo":{"doubles":memo},"search_attributes":attributes})).await;
+    assert_eq!(status, StatusCode::CREATED, "{started}");
+    let task = poll(&app, "doubles-worker", "workflow").await;
+    let (status, body) = finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"measure",
+        "arguments":envelope(Payload::Array(vec![])),"start_to_close_timeout":300}])).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let activity = poll(&app, "doubles-worker", "activity").await;
+    let (status, body) = request(&app, "POST", &format!("/api/worker/activity-tasks/{}/heartbeat", activity["task_id"].as_str().unwrap()),
+        json!({"activity_attempt_id":activity["activity_attempt_id"],"lease_owner":activity["lease_owner"],"details":memo})).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+
+    // Read everything back through a fresh runtime, from storage alone.
+    runtime.close().await;
+    drop(app);
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let run = format!("/api/workflows/doubles/runs/{}", started["run_id"].as_str().unwrap());
+    let described = request(&app, "GET", &run, Value::Null).await.1;
+    let history = run_history(&app, &json!("doubles"), &started["run_id"]).await;
+    let workflow_started = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowStarted").unwrap()["payload"].clone();
+    let listed = request(&app, "GET", &format!("/api/workflows?query={}", "WorkflowId%20%3D%20%22doubles%22"), Value::Null).await.1;
+    let activities = request(&app, "GET", &format!("{run}/activities"), Value::Null).await.1;
+    let heartbeat = &activities["activities"][0]["last_heartbeat_progress"]["details"];
+    for (index, text) in DOUBLES.iter().enumerate() {
+        let (key, name) = (format!("d{index}"), format!("Dbl{index}"));
+        exact(&described["memo"]["doubles"], &key, text, "described memo");
+        exact(&workflow_started["memo"]["doubles"], &key, text, "WorkflowStarted memo");
+        exact(&described["search_attributes"], &name, text, "described search attribute");
+        exact(&listed["workflows"][0]["search_attributes"], &name, text, "listed search attribute");
+        exact(&workflow_started["search_attributes"], &name, text, "WorkflowStarted search attribute");
+        exact(heartbeat, &key, text, "heartbeat details");
+    }
+    runtime.close().await;
+    database.remove().await;
+}

rust/tests/execution/memos.rs

@@ -533,18 +533,6 @@ async fn raw_text(app: &Router, method: &str, path: &str, body: &str) -> (Status
     (status, String::from_utf8(bytes.to_vec()).unwrap())
 }
 
-/// The double a backend's JSON column holds for `text`. MySQL's json type
-/// parses doubles itself and can land one ULP off, for PHP and native alike;
-/// MariaDB stores JSON as text, and SQLite and PostgreSQL keep the text.
-async fn json_column_double(database: &TestDatabase, text: &str) -> f64 {
-    if !matches!(database, TestDatabase::MySql { .. }) {
-        return text.parse().unwrap();
-    }
-    let query = format!(r#"SELECT 'value', CAST(JSON_EXTRACT('{{"x":{text}}}', '$.x') AS CHAR)"#);
-    let rows = text_rows(database, "", "", &query).await;
-    rows[0].1.parse().unwrap()
-}
-
 // PHP's json_decode and json_encode give a double back with the digits it
 // came in with. Without serde_json's float_roundtrip, parsing can land one
 // ULP off, so a memo's 1.4820142857142857 was answered as …855.
@@ -575,10 +563,8 @@ async fn memo_doubles_keep_every_digit_through_storage() {
         let exact: f64 = text.parse().unwrap();
         let printed = serde_json::to_string(&exact).unwrap();
         assert!(described.contains(&format!("\"d{index}\":{printed}")), "describe d{index} {text}: {described}");
-        // History payloads live in a JSON column, so they hold the double
-        // that column stores, which the database itself names.
-        let stored = serde_json::to_string(&json_column_double(&database, text).await).unwrap();
-        assert!(history.contains(&format!("\"d{index}\":{stored}")), "history d{index} {text} as {stored}: {history}");
+        // History keeps every digit too, MySQL included (server#1016).
+        assert!(history.contains(&format!("\"d{index}\":{printed}")), "history d{index} {text}: {history}");
     }
     runtime.close().await;
     database.remove().await;

rust/tests/execution/query_contract_arguments.rs

@@ -122,22 +122,15 @@ async fn a_registered_json_default_takes_phps_array_shape() {
     start(&app, "contract-query").await;
     let list = Payload::Array(vec![Payload::String("x".into()), Payload::String("y".into())]);
     assert_eq!(routed(&app, "listdef", envelope(Payload::Array(vec![]))).await, envelope(Payload::Array(vec![list.clone()])));
-    // A map default reaches the worker in the key order the database keeps:
-    // the registered order, except that a MySQL json column orders keys by
-    // length, then bytes, for PHP as for native. MariaDB's json is text.
-    let version = text_rows(&database, "SELECT 'version', sqlite_version()", "SELECT 'version'::text, version()",
-        "SELECT 'version', VERSION()").await;
-    let mysql = matches!(database, TestDatabase::MySql { .. }) && !version[0].1.contains("MariaDB");
+    // A map default reaches the worker in its registered key order on every
+    // backend. MySQL's json column used to order keys by length, then bytes;
+    // native schema 13 stores JSON as text there too (server#1016).
     let (z, a) = (("z".to_owned(), OrderedValue::Long(20)), ("a".to_owned(), OrderedValue::Long(10)));
-    let ordered = OrderedValue::Array(vec![OrderedValue::Map(if mysql { vec![a, z] } else { vec![z, a] })]);
+    let ordered = OrderedValue::Array(vec![OrderedValue::Map(vec![z, a])]);
     assert_eq!(routed(&app, "orderdef", envelope(Payload::Array(vec![]))).await,
         json!({"codec": "avro", "blob": ValueCodec::new().unwrap().encode_ordered_value(&ordered).unwrap()}));
     // Keys 0 to n-1 out of order are not a list, so PHP's encoder fails.
-    // MySQL stores them in order, which makes them a list.
-    if mysql {
-        assert_eq!(routed(&app, "descdef", envelope(Payload::Array(vec![]))).await, envelope(Payload::Array(vec![list.clone()])));
-    }
-    for query in if mysql { vec!["gapdef"] } else { vec!["gapdef", "descdef"] } {
+    for query in ["gapdef", "descdef"] {
         let (status, body) = request(&app, "POST", &format!("/api/workflows/contract-query/query/{query}"),
             json!({"input": envelope(Payload::Array(vec![]))})).await;
         assert_eq!((status, body["reason"].clone()), (StatusCode::INTERNAL_SERVER_ERROR, json!("query_default_unencodable")),

scripts/conformance/server-parity/json-doubles-contract.mjs

@@ -7,10 +7,11 @@ import assert from 'node:assert/strict';
 // - published PHP renders a typed double search attribute through PHP's
 //   14-digit `precision` in the run description and the visibility list
 //   (`php_search_attribute_precision`); native keeps every digit;
-// - on MySQL, the json column type parses doubles itself, so history
-//   payloads (WorkflowStarted, heartbeat details) hold MySQL's double for
-//   both implementations (`mysql_json_doubles`, read from MySQL 8.0 with
-//   CAST(... AS JSON)). MariaDB, SQLite and PostgreSQL keep the text.
+// - on MySQL, published PHP's json column type parses doubles itself, so
+//   its history payloads (WorkflowStarted, heartbeat details) hold MySQL's
+//   double (`mysql_json_doubles`, read from MySQL 8.0 with CAST(... AS
+//   JSON)). Native stores JSON as exact text from schema 13 (server#1016),
+//   as MariaDB, SQLite and PostgreSQL keep it.
 // Expectations come from the fixture's texts, never from a server recording.
 const HISTORY_READERS = ['started_memo', 'started_search_attributes', 'heartbeat_details'];
 const ATTRIBUTE_READERS = ['described_search_attributes', 'listed_search_attributes'];
@@ -39,7 +40,8 @@ export function checkJsonDoubles(fixture, observation) {
     assert.ok(Number.isFinite(value) && !Number.isInteger(value), `fixture double ${name} has a fraction`);
     return value;
   };
-  const stored = (name) => storage === 'mysql_json' && Object.hasOwn(definition.mysql_json_doubles, name)
+  const stored = (name) => storage === 'mysql_json' && implementation === null
+    && Object.hasOwn(definition.mysql_json_doubles, name)
     ? Number(definition.mysql_json_doubles[name]) : exact(name);
   const attribute = (name) => implementation === null
     ? Number(exact(name).toPrecision(definition.php_search_attribute_precision)) : exact(name);

scripts/conformance/server-parity/mysql-schema.php

@@ -108,7 +108,7 @@ function catalog(PDO $pdo): array
 }
 $pdo->exec('ROLLBACK');
 $pdo = connection($argv[3]);
-$native = array_values(array_filter(projectMysqlScopedDispatch(projectMysqlReadyOrigin(catalog($pdo))), static fn (array $row): bool => ! in_array($row[0], ['dw_server_schema', '_sqlx_migrations', 'dw_task_completions', 'dw_poll_receipts', 'dw_query_cache', 'dw_worker_registration_incarnations', 'dw_worker_session_holders', 'dw_stream_reclaims', 'dw_service_call_deadlines', 'dw_service_admission_counters', 'dw_external_payload_deletions', 'dw_run_command_contracts'], true)));
+$native = array_values(array_filter(projectMysqlJsonText(projectMysqlScopedDispatch(projectMysqlReadyOrigin(catalog($pdo))), $php), static fn (array $row): bool => ! in_array($row[0], ['dw_server_schema', '_sqlx_migrations', 'dw_task_completions', 'dw_poll_receipts', 'dw_query_cache', 'dw_worker_registration_incarnations', 'dw_worker_session_holders', 'dw_stream_reclaims', 'dw_service_call_deadlines', 'dw_service_admission_counters', 'dw_external_payload_deletions', 'dw_run_command_contracts'], true)));
 $pdo->exec('ROLLBACK');
 // Native polling uses an additional index on an existing table. Retain all
 // original column, index and constraint definitions in the comparison.

scripts/conformance/server-parity/ready-origin-schema.php

@@ -108,3 +108,35 @@ function projectPostgresScopedDispatch(array $rows): array
 
     return $result;
 }
+
+// Native MySQL schema 13 stores each json column as exact utf8mb4_bin text
+// with an enforced json_valid check (server#1016). For every json column in
+// published PHP's table, map exactly that native definition and its named
+// check back to PHP's json definition. Any other difference still fails.
+function projectMysqlJsonText(array $native, array $php): array
+{
+    $source = array_column($php, 1, 0);
+    foreach ($native as &$row) {
+        [$table, $sql] = $row;
+        if (! isset($source[$table])) {
+            continue;
+        }
+        preg_match_all("/^  `(\\w+)` json( NOT NULL| DEFAULT NULL)( COMMENT '[^']*')?,$/m", $source[$table], $columns, PREG_SET_ORDER);
+        foreach ($columns as $column) {
+            [$line, $name, $null] = $column;
+            $comment = $column[3] ?? '';
+            $pattern = '/^  `'.preg_quote($name, '/').'` longtext( CHARACTER SET utf8mb4)? COLLATE utf8mb4_bin'
+                .($null === ' NOT NULL' ? ' NOT NULL' : '').preg_quote($comment, '/').',$/m';
+            $sql = preg_replace($pattern, $line, $sql, -1, $replaced);
+            $check = ",\n  CONSTRAINT `dw_json_text_".substr(hash('sha256', $table."\0".$name), 0, 20)."` CHECK (json_valid(`{$name}`))";
+            if ($replaced !== 1 || substr_count($sql, $check) !== 1) {
+                throw new RuntimeException("Native JSON text definition mismatch: {$table}.{$name}");
+            }
+            $sql = str_replace($check, '', $sql);
+        }
+        $row[1] = $sql;
+    }
+    unset($row);
+
+    return $native;
+}