fix(2meet-data-optimizer): alter an existing flat table in one process at a time; keep a deferred group native until its drift fix

- security review H1: anonymous /wp-cron.php requests pass ddl_allowed(),
  and dbDelta on an existing table had no lock, so concurrent requests each
  ran the ALTER on a large table. The upgrade now runs under the same MySQL
  GET_LOCK as the index DDL (skipped when another process holds it) and
  re-checks the stored hash once locked; this covers aeav_only too. The table
  build moves unchanged to write_table(); lock()/unlock() are shared
- security review M1: after a deferred upgrade the group stays native until
  tmdo_drift_resync has rebuilt the rows written meanwhile
  (TMDO_Schema_Manager::resynced() from run_scheduled_fix()); with no flat
  writes (disabled) it is released at once
- Verify: SchemaUpgradeDeferralTest (lock held elsewhere, resync release),
  integration 585, unit 702, phpcs, stan
- PR: PR-111 (2.3 acceptance security review)

cli/class-tmdo-cli.php

@@ -144,7 +144,7 @@ class TMDO_CLI {
 		}
 		$deferred = array_keys( (array) get_option( TMDO_Schema_Manager::DEFERRED_OPTION, array() ) );
 		if ( $deferred ) {
-			WP_CLI::warning( '  [SCHEMA] table upgrades not done yet (these groups use native meta meanwhile): ' . implode( ', ', $deferred ) . ' — see the error log if this run did not finish them.' );
+			WP_CLI::warning( '  [SCHEMA] groups using native meta until their table upgrade and the drift resync after it (tmdo_drift_resync) are done: ' . implode( ', ', $deferred ) . '.' );
 		}
 		$pending = TMDO_Schema_Manager::pending_indexes();
 		if ( $pending ) {

includes/diagnostic/class-tmdo-drift-check.php

@@ -237,6 +237,7 @@ final class TMDO_Drift_Check {
 	public static function run_scheduled_fix( string $entity_type ): void {
 		try {
 			$result = self::fix( $entity_type );
+			TMDO_Schema_Manager::resynced( $entity_type ); // Groups upgraded while deferred use the flat table again.
 			TMDO_Logger::info( 'drift_resync', array( 'entity' => $entity_type ) + $result['fixed'] );
 		} catch ( InvalidArgumentException | RuntimeException $e ) {
 			TMDO_Logger::info(

includes/engine/class-tmdo-schema-manager.php

@@ -83,8 +83,6 @@ final class TMDO_Schema_Manager {
 		string $group_name,
 		array $field_definitions
 	): bool {
-		global $wpdb;
-
 		// Schema 版本比對：若未變動則跳過
 		$schema_hash = self::calculate_schema_hash( $field_definitions );
 		$stored_hash = self::get_stored_schema_hash( $entity_type, $group_name );
@@ -103,6 +101,39 @@ final class TMDO_Schema_Manager {
 			self::defer_upgrade( "{$entity_type}/{$group_name}" );
 			return false;
 		}
+		if ( ! self::table_exists( $table ) ) {
+			return self::write_table( $entity_type, $group_name, $field_definitions, $adapter, $table, $schema_hash );
+		}
+
+		// An existing table is altered by one process at a time (anonymous
+		// /wp-cron.php requests also pass ddl_allowed()); a process that waited
+		// finds the upgrade done.
+		if ( ! self::lock() ) {
+			return false;
+		}
+		try {
+			if ( self::get_stored_schema_hash( $entity_type, $group_name ) === $schema_hash ) {
+				return true;
+			}
+			return self::write_table( $entity_type, $group_name, $field_definitions, $adapter, $table, $schema_hash );
+		} finally {
+			self::unlock();
+		}
+	}
+
+	/**
+	 * Create or alter the table (dbDelta) and record its definition.
+	 *
+	 * @param string                           $entity_type       Entity type.
+	 * @param string                           $group_name        Group name.
+	 * @param array<int, array<string, mixed>> $field_definitions Normalized fields.
+	 * @param TMDO_Entity_Adapter_Interface    $adapter           Entity adapter.
+	 * @param string                           $table             Table name.
+	 * @param string                           $schema_hash       Definition hash.
+	 * @return bool
+	 */
+	private static function write_table( string $entity_type, string $group_name, array $field_definitions, TMDO_Entity_Adapter_Interface $adapter, string $table, string $schema_hash ): bool {
+		global $wpdb;
 
 		$charset = $wpdb->get_charset_collate();
 		$id_col  = $adapter->get_entity_id_column();
@@ -227,16 +258,7 @@ final class TMDO_Schema_Manager {
 			return;
 		}
 
-		// add_option() is INSERT ... ON DUPLICATE KEY UPDATE and would let two
+		if ( ! self::lock() ) {
-		// processes both "acquire" an option lock; GET_LOCK is atomic and is
-		// released by MySQL if this process dies.
-		$lock = self::index_lock_name();
-		if ( method_exists( $wpdb, 'send_reads_to_masters' ) ) {
-			// HyperDB / LudicrousDB would send the SELECT GET_LOCK to a replica,
-			// where it locks nothing against the ALTER on the primary.
-			$wpdb->send_reads_to_masters();
-		}
-		if ( '1' !== (string) $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK(MD5(CONCAT(DATABASE(), %s)), 0)', $lock ) ) ) {
 			return; // Another process is adding indexes.
 		}
 		try {
@@ -267,7 +289,7 @@ final class TMDO_Schema_Manager {
 				update_option( self::INDEXES_OPTION, $applied, true );
 			}
 		} finally {
-			$wpdb->query( $wpdb->prepare( 'SELECT RELEASE_LOCK(MD5(CONCAT(DATABASE(), %s)))', $lock ) );
+			self::unlock();
 		}
 	}
 
@@ -331,8 +353,10 @@ final class TMDO_Schema_Manager {
 	}
 
 	/**
-	 * A deferred upgrade is done: writes made meanwhile skipped the flat table,
+	 * A deferred upgrade is done. Writes made meanwhile skipped the flat table,
-	 * so rebuild it from native (drift fix) a minute later.
+	 * so the group stays native until a drift fix (tmdo_drift_resync, a minute
+	 * later) has rebuilt it from native and calls resynced(). With no flat
+	 * writes (disabled), nothing is behind.
 	 *
 	 * @param string $entity_type Entity type.
 	 * @param string $group_name  Group name.
@@ -342,14 +366,67 @@ final class TMDO_Schema_Manager {
 		if ( ! self::is_deferred( $entity_type, $group_name ) ) {
 			return;
 		}
+		if ( ! TMDO_Mode_Manager::writes_to_flat( $entity_type ) ) {
 			$deferred = (array) get_option( self::DEFERRED_OPTION, array() );
 			unset( $deferred[ "{$entity_type}/{$group_name}" ] );
 			update_option( self::DEFERRED_OPTION, $deferred, true );
-		if ( TMDO_Mode_Manager::writes_to_flat( $entity_type ) && ! wp_next_scheduled( TMDO_Drift_Check::RESYNC_HOOK, array( $entity_type ) ) ) {
+			return;
+		}
+		if ( ! wp_next_scheduled( TMDO_Drift_Check::RESYNC_HOOK, array( $entity_type ) ) ) {
 			wp_schedule_single_event( time() + MINUTE_IN_SECONDS, TMDO_Drift_Check::RESYNC_HOOK, array( $entity_type ) );
 		}
 	}
 
+	/**
+	 * A drift fix of the entity succeeded: its deferred groups whose table is
+	 * upgraded read and write the flat table again.
+	 *
+	 * @param string $entity_type Entity type.
+	 * @return void
+	 */
+	public static function resynced( string $entity_type ): void {
+		$deferred = get_option( self::DEFERRED_OPTION, array() );
+		if ( ! is_array( $deferred ) || ! $deferred ) {
+			return;
+		}
+		foreach ( array_keys( $deferred ) as $id ) {
+			list( $type, $group ) = array_pad( explode( '/', (string) $id, 2 ), 2, '' );
+			if ( $type === $entity_type && self::get_stored_schema_hash( $type, $group ) === self::calculate_schema_hash( TMDO_Entity_Registry::get_group_fields( $type, $group ) ) ) {
+				unset( $deferred[ $id ] );
+			}
+		}
+		update_option( self::DEFERRED_OPTION, $deferred, true );
+	}
+
+	/**
+	 * Take the schema lock without waiting.
+	 *
+	 * add_option() is INSERT ... ON DUPLICATE KEY UPDATE and would let two
+	 * processes both "acquire" an option lock; GET_LOCK is atomic and is
+	 * released by MySQL if this process dies. Re-entrant per connection.
+	 *
+	 * @return bool False when another process holds it.
+	 */
+	private static function lock(): bool {
+		global $wpdb;
+		if ( method_exists( $wpdb, 'send_reads_to_masters' ) ) {
+			// HyperDB / LudicrousDB would send the SELECT GET_LOCK to a replica,
+			// where it locks nothing against the ALTER on the primary.
+			$wpdb->send_reads_to_masters();
+		}
+		return '1' === (string) $wpdb->get_var( $wpdb->prepare( 'SELECT GET_LOCK(MD5(CONCAT(DATABASE(), %s)), 0)', self::index_lock_name() ) );
+	}
+
+	/**
+	 * Release the schema lock.
+	 *
+	 * @return void
+	 */
+	private static function unlock(): void {
+		global $wpdb;
+		$wpdb->query( $wpdb->prepare( 'SELECT RELEASE_LOCK(MD5(CONCAT(DATABASE(), %s)))', self::index_lock_name() ) );
+	}
+
 	/**
 	 * Key of the MySQL lock held while indexes are added. Lock names are
 	 * server-wide, so SQL hashes it with DATABASE() to keep sites apart.

tests/integration/SchemaUpgradeDeferralTest.php

@@ -88,4 +88,55 @@ class SchemaUpgradeDeferralTest extends TestCase {
 		$this->assertFalse( TMDO_Schema_Manager::upgrade_waits( 'user', self::TABLE ) );
 		$GLOBALS['_wp_filter_callbacks']['tmdo_schema_ddl_allowed'] = array();
 	}
+
+	/**
+	 * An existing table is altered by one process at a time: anonymous
+	 * /wp-cron.php requests pass ddl_allowed() too (security review H1).
+	 */
+	public function test_an_upgrade_is_skipped_while_another_process_holds_the_schema_lock(): void {
+		add_filter( 'tmdo_schema_ddl_allowed', static fn() => true );
+		$other = new mysqli(
+			getenv( 'TMDO_TEST_DB_HOST' ) ?: '127.0.0.1',
+			getenv( 'TMDO_TEST_DB_USER' ) ?: 'dbo',
+			(string) getenv( 'TMDO_TEST_DB_PASS' ),
+			getenv( 'TMDO_TEST_DB_NAME' ) ?: 'wp_wpdo_test'
+		);
+		$name = $other->real_escape_string( TMDO_Schema_Manager::index_lock_name() );
+		$other->query( "SELECT GET_LOCK(MD5(CONCAT(DATABASE(), '{$name}')), 0)" );
+		try {
+			$this->assertFalse( $this->upgrade(), 'returns before dbDelta (unavailable here)' );
+		} finally {
+			$other->query( "SELECT RELEASE_LOCK(MD5(CONCAT(DATABASE(), '{$name}')))" );
+			$other->close();
+			$GLOBALS['_wp_filter_callbacks']['tmdo_schema_ddl_allowed'] = array();
+		}
+		$this->assertFalse( TMDO_Schema_Manager::is_deferred( 'user', 'sd_core' ) );
+	}
+
+	/**
+	 * After the upgrade the group stays native until the drift fix has
+	 * rebuilt the rows written meanwhile (security review M1).
+	 */
+	public function test_a_drift_fix_releases_only_upgraded_groups_of_its_entity(): void {
+		global $wpdb;
+		$current = TMDO_Schema_Manager::calculate_schema_hash( WPDO_Entity_Registry::get_group_fields( 'user', 'sd_core' ) );
+		update_option(
+			TMDO_Schema_Manager::DEFERRED_OPTION,
+			array(
+				'user/sd_core'  => 1,
+				'user/sd_later' => 1, // Not upgraded yet (no stored definition).
+				'post/sd_core'  => 1, // Another entity.
+			)
+		);
+
+		TMDO_Schema_Manager::resynced( 'user' );
+		$this->assertTrue( TMDO_Schema_Manager::is_deferred( 'user', 'sd_core' ), 'stored hash still old: not upgraded' );
+
+		$wpdb->update( $wpdb->prefix . 'wpdo_registry_meta', array( 'schema_hash' => $current ), array( 'entity_type' => 'user', 'group_name' => 'sd_core' ) );
+		TMDO_Schema_Manager::resynced( 'user' );
+
+		$this->assertFalse( TMDO_Schema_Manager::is_deferred( 'user', 'sd_core' ) );
+		$this->assertTrue( TMDO_Schema_Manager::is_deferred( 'user', 'sd_later' ) );
+		$this->assertTrue( TMDO_Schema_Manager::is_deferred( 'post', 'sd_core' ) );
+	}
 }