← All changes | includes/Sync/Integrity_Digest.php +243 -27 1.10.4 → 1.10.17 View file →

		@@ -17,16 +17,20 @@
17	17
18	18	/**
19	19	* Hash-backed range-checksum support: stored per-record content digests.
20	20	*
21		- * STORES a digest of each product/variation's raw DB row at hook time (the
22		- * same save/delete hooks class-change-log.php uses), so the integrity scan
	21	+ * STORES a digest of each product/variation's raw DB row, marked dirty by the
	22	+ * same save/delete hooks class-change-log.php uses and written at the request
	23	+ * boundary (see $pending_digests), so the integrity scan
23	24	* can compare — entirely in SQL — the aggregate of CURRENT raw-row digests
24	25	* against the aggregate of STORED digests per id-range bucket. If hooks
25	26	* fired for every write, stored == current (and sequence-log already
26	27	* reported the change); a bucket mismatch therefore means exactly "content
27	28	* changed without hooks firing" — the sql-bypass signature — at GROUP BY
28		- * prices instead of revision-hash's full-hydration prices.
	29	+ * prices instead of revision-hash's full-hydration prices. Because the write
	30	+ * lands at the boundary, a hook-less write later in the SAME request is
	31	+ * absorbed into that request's digest; the scan catches bypasses between
	32	+ * requests, which is where they happen (a direct SQL job, an importer).
29	33	*
30	34	* The digest basis is deliberately the RAW DB ROW, NOT the filtered REST
31	35	* payload: this signal is detection-only (discovery of WHERE drift
32	36	* happened, ADR 0003 "discovery, never values"); hydration of anything the
		@@ -76,8 +80,51 @@
76	80	* computed by ONE expression — the invariant the whole scan rests on.
77	81	*/
78	82	private Digest_Index $index;
79	83
	84	+ /**
	85	+ * Order and customer digests owed but not yet written, keyed "type:id".
	86	+ *
	87	+ * A stored digest is a pure function of the settled record, so only the
	88	+ * LAST upsert in a request carries information — yet one Store API checkout
	89	+ * ran the order INSERT…SELECT eleven times (35 ms) and, with account
	90	+ * creation, the customer one six more times (measured 2026-09-03 on
	91	+ * dev-next). Upserts land on {@see flush_pending_digests()}: at `shutdown`
	92	+ * (last, after WooCommerce's own customer save at 10 and session save at 20,
	93	+ * whose `woocommerce_update_customer` would otherwise queue after the only
	94	+ * flush), before any {@see Digest_Index::read_digests()} so the pull lane
	95	+ * never stamps a stale `_rxdb_digest`, and whenever the queue reaches
	96	+ * {@see PENDING_DIGEST_FLUSH_THRESHOLD} distinct records (a bulk import
	97	+ * coalesces nothing, so it must not accumulate). Once the shutdown flush
	98	+ * has run, later saves write immediately. Static so the read path can
	99	+ * flush without holding the observer instance; each entry keeps its blog id
	100	+ * so a multisite `switch_to_blog()` between save and flush still writes the
	101	+ * originating site's table. Product and variation digests ride the same
	102	+ * queue: WooCommerce saves a product more than once per request too
	103	+ * (`wc_reduce_stock_levels()` saves the quantity, then the stock status —
	104	+ * two INSERT…SELECT statements per purchased product at checkout, measured
	105	+ * 2026-09-03), and the v2 write lane reads digests back through
	106	+ * {@see Digest_Index::read_digests()}, which flushes first, so the
	107	+ * serializer still stamps the fresh `_rxdb_digest` in the same request.
	108	+ *
	109	+ * @var array<string, array{0: int, 1: string, 2: int}> "blog:type:id" => [blog, type, id]
	110	+ */
	111	+ private static array $pending_digests = array();
	112	+
	113	+ /**
	114	+ * Flush the queue when it holds this many distinct records. Sized for the
	115	+ * realistic per-request maximum (a checkout touches an order and a customer;
	116	+ * a REST batch a few dozen records) while keeping a WP-CLI import's deferred
	117	+ * SQL and memory bounded.
	118	+ */
	119	+ public const PENDING_DIGEST_FLUSH_THRESHOLD = 50;
	120	+
	121	+ /** The instance that first queued a digest; the flush writes through its Digest_Index. */
	122	+ private static ?Integrity_Digest $flusher = null;
	123	+
	124	+ /** Set by the shutdown flush; afterwards saves are written immediately. */
	125	+ private static bool $shutdown_flushed = false;
	126	+
80	127	public function __construct( ?Digest_Index $index = null ) {
81	128	$this->index = $index ?? new Digest_Index();
82	129	}
83	130
		@@ -160,8 +207,13 @@
160	207	// never fires for COT orders — without the HPOS twin hook a restored
161	208	// order's digest is never recreated and integrity scans treat it as
162	209	// deleted forever.
163	210	add_action( 'woocommerce_untrash_order', array( $this, 'record_order_untrashed' ), 10, 1 );
	211	+ // Request boundary for the coalesced digest upserts (see
	212	+ // $pending_digests). LAST on shutdown: WooCommerce saves the customer at
	213	+ // 10 and the session at 20. Zero accepted args: do_action( 'shutdown' )
	214	+ // passes an empty string otherwise.
	215	+ add_action( 'shutdown', array( __CLASS__, 'flush_pending_digests_at_shutdown' ), PHP_INT_MAX, 0 );
164	216	}
165	217
166	218	/**
167	219	* Recreate a COT order's digest once its restore completes.
		@@ -242,8 +294,23 @@
242	294	return $registered;
243	295	}
244	296
245	297	/**
	298	+ * Detach THE digest stamper from both served read lanes.
	299	+ *
	300	+ * The teardown twin of {@see register_proxy_digest_stampers()}, matching the
	301	+ * `unregister_*` seams {@see Revision} and {@see Proxy_Uuid_Stamper} already
	302	+ * expose. `Augmentation_Pipeline::reset()` only removes the projections the
	303	+ * pipeline itself installed, so without this a caller that installs the real
	304	+ * pipeline — a test wiring the production read lane — cannot unwind it and
	305	+ * leaks this filter into everything that runs after it.
	306	+ */
	307	+ public static function unregister_proxy_digest_stampers(): void {
	308	+ remove_filter( 'woocommerce_pos_sync_proxy_response', array( __CLASS__, 'stamp_digests' ), 10 );
	309	+ remove_filter( 'woocommerce_pos_sync_order_pull_payloads', array( __CLASS__, 'stamp_digests' ), 10 );
	310	+ }
	311	+
	312	+ /**
246	313	* Attach each served record's stored 64-bit digest as a top-level `_rxdb_digest`
247	314	* string, so the client seeds its existence-reconcile manifest (ADR 0014 Leg 3)
248	315	* as records flow through the NORMAL pull — no separate fetch. The client reads
249	316	* it into the sidecar manifest; it is NOT persisted into the document. A record
		@@ -298,17 +365,113 @@
298	365	/**
299	366	* Customer digest maintenance (ADR 0015, Leg-3 phase 7) — every WordPress
300	367	* user is a POS customer, so saves and role changes always upsert.
301	368	*/
	369	+ /** Owe the customer's digest; it is written once, on flush (see $pending_digests). */
302	370	public function record_customer_saved( int $user_id ): void {
	371	+ $this->defer( 'customer', $user_id );
	372	+ }
	373	+
	374	+ /**
	375	+ * Queue one digest upsert, or write it now if the boundary has passed.
	376	+ *
	377	+ * @param string $type 'order' or 'customer'.
	378	+ * @param int $id Record id.
	379	+ */
	380	+ private function defer( string $type, int $id ): void {
	381	+ if ( self::$shutdown_flushed ) {
	382	+ // A save triggered by another shutdown handler (WooCommerce saves the
	383	+ // customer at priority 10): nothing will flush again, so write now.
	384	+ $this->upsert_pending( $type, $id );
	385	+ return;
	386	+ }
	387	+ if ( null === self::$flusher ) {
	388	+ self::$flusher = $this;
	389	+ }
	390	+ $blog = get_current_blog_id();
	391	+ self::$pending_digests[ self::pending_key( $type, $id ) ] = array( $blog, $type, $id );
	392	+ if ( \count( self::$pending_digests ) >= self::PENDING_DIGEST_FLUSH_THRESHOLD ) {
	393	+ self::flush_pending_digests();
	394	+ }
	395	+ }
	396	+
	397	+ private static function pending_key( string $type, int $id ): string {
	398	+ return get_current_blog_id() . ':' . $type . ':' . $id;
	399	+ }
	400	+
	401	+ /** One queued upsert, under the observer's fail-open posture. */
	402	+ private function upsert_pending( string $type, int $id ): void {
303	403	$this->observe(
304		- function () use ( $user_id ): void {
305		- $this->upsert_customer_digest( $user_id );
	404	+ function () use ( $type, $id ): void {
	405	+ if ( 'customer' === $type ) {
	406	+ $this->upsert_customer_digest( $id );
	407	+ } elseif ( 'order' === $type ) {
	408	+ $this->upsert_order_digest( $id );
	409	+ } else {
	410	+ // 'post' (product or variation): the SQL derives the stored type from the row.
	411	+ $this->upsert_digest( $id );
	412	+ }
306	413	}
307	414	);
308	415	}
309	416
	417	+ /**
	418	+ * Write every owed digest.
	419	+ *
	420	+ * Called from the shutdown flush, from {@see Digest_Index::read_digests()}
	421	+ * before it reads, and when the queue reaches its threshold. Writes go
	422	+ * through the instance that first queued (so an injected Digest_Index is
	423	+ * honoured) and under the blog each entry was recorded on. Each upsert keeps
	424	+ * the observer's fail-open posture: a failure is logged and the scan
	425	+ * self-heals. Safe to call repeatedly — a flushed digest is no longer pending.
	426	+ */
	427	+ public static function flush_pending_digests(): void {
	428	+ if ( array() === self::$pending_digests ) {
	429	+ return;
	430	+ }
	431	+ $pending = self::$pending_digests;
	432	+ self::$pending_digests = array();
	433	+ $digest = self::$flusher ?? new self();
	434	+ foreach ( $pending as $entry ) {
	435	+ list( $blog, $type, $id ) = $entry;
	436	+ $switch = is_multisite() && get_current_blog_id() !== (int) $blog;
	437	+ if ( $switch ) {
	438	+ switch_to_blog( (int) $blog );
	439	+ }
	440	+ try {
	441	+ $digest->upsert_pending( (string) $type, (int) $id );
	442	+ } finally {
	443	+ if ( $switch ) {
	444	+ restore_current_blog();
	445	+ }
	446	+ }
	447	+ }
	448	+ }
	449	+
	450	+ /**
	451	+ * The `shutdown` callback: flush, then write every later save immediately.
	452	+ */
	453	+ public static function flush_pending_digests_at_shutdown(): void {
	454	+ self::$shutdown_flushed = true;
	455	+ self::flush_pending_digests();
	456	+ }
	457	+
	458	+ /**
	459	+ * Discard per-request coalescing state. Tests only: the PHPUnit process
	460	+ * never reaches `shutdown`, so the static queue, flusher and flag would
	461	+ * leak between test cases otherwise.
	462	+ *
	463	+ * @internal
	464	+ */
	465	+ public static function reset_request_state(): void {
	466	+ self::$pending_digests = array();
	467	+ self::$flusher = null;
	468	+ self::$shutdown_flushed = false;
	469	+ }
	470	+
310	471	public function record_customer_deleted( int $user_id ): void {
	472	+ // A pending upsert for a record that is leaving must not be written after the fact.
	473	+ unset( self::$pending_digests[ self::pending_key( 'customer', $user_id ) ] );
311	474	$this->observe(
312	475	function () use ( $user_id ): void {
313	476	$this->delete_customer_digest( $user_id );
314	477	}
		@@ -351,17 +514,16 @@
351	514	* Order digest maintenance (ADR 0015, Leg-3 phase 7). The WC order hooks are storage-agnostic (fire
352	515	* under HPOS AND CPT); the digest SQL's `type='shop_order'` filter makes the upsert a no-op for any
353	516	* non-order, so no type re-check is needed here.
354	517	*/
	518	+ /** Owe the order's digest; it is written once, on flush (see $pending_digests). */
355	519	public function record_order_saved( int $order_id ): void {
356		- $this->observe(
357		- function () use ( $order_id ): void {
358		- $this->upsert_order_digest( $order_id );
359		- }
360		- );
	520	+ $this->defer( 'order', $order_id );
361	521	}
362	522
363	523	public function record_order_deleted( int $order_id ): void {
	524	+ // A pending upsert for a record that is leaving must not be written after the fact.
	525	+ unset( self::$pending_digests[ self::pending_key( 'order', $order_id ) ] );
364	526	$this->observe(
365	527	function () use ( $order_id ): void {
366	528	$this->delete_order_digest( $order_id );
367	529	}
		@@ -402,22 +564,26 @@
402	564	throw new RuntimeException( 'upsert stored order digest failed: ' . $wpdb->last_error );
403	565	}
404	566	}
405	567
	568	+ /**
	569	+ * Owe the product's or variation's digest; it is written once, on flush (see
	570	+ * $pending_digests). The queue type is 'post' for both: the upsert's SQL
	571	+ * derives the stored object_type from the row, so nothing here needs to.
	572	+ */
406	573	public function record_post_saved( int $post_id ): void {
407		- $this->observe(
408		- function () use ( $post_id ): void {
409		- $this->upsert_digest( $post_id );
410		- }
411		- );
	574	+ $this->defer( 'post', $post_id );
412	575	}
413	576
414	577	public function record_post_untrashed( int $post_id ): void {
415		- if ( 'shop_order' === get_post_type( $post_id ) ) {
	578	+ $post_type = get_post_type( $post_id );
	579	+ if ( 'shop_order' === $post_type ) {
416	580	$this->record_order_saved( $post_id );
417	581	return;
418	582	}
419		- $this->record_post_saved( $post_id );
	583	+ if ( in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
	584	+ $this->record_post_saved( $post_id );
	585	+ }
420	586	}
421	587
422	588	public function record_post_deleted( int $post_id ): void {
423	589	$post_type = get_post_type( $post_id );
		@@ -423,8 +589,10 @@
423	589	$post_type = get_post_type( $post_id );
424	590	if ( ! in_array( $post_type, array( 'product', 'product_variation' ), true ) ) {
425	591	return;
426	592	}
	593	+ // A pending upsert for a record that is leaving must not be written after the fact.
	594	+ unset( self::$pending_digests[ self::pending_key( 'post', $post_id ) ] );
427	595	$this->observe(
428	596	function () use ( $post_id, $post_type ): void {
429	597	$this->delete_post_digest( $post_id, $post_type );
430	598	}
		@@ -459,9 +627,9 @@
459	627	}
460	628	}
461	629
462	630	/**
463		- * One round trip: the digest is computed in SQL from the raw row and
	631	+ * One statement: the digest is computed in SQL from the raw row and
464	632	* upserted in the same statement — PHP never materializes the value.
465	633	* No-op for rows outside the live predicate (the delete hook owns those).
466	634	*/
467	635	public function upsert_digest( int $post_id ): void {
		@@ -469,9 +637,9 @@
469	637	// Time from BEFORE the session setup so timing.digest_ms covers ALL digest hook work
470	638	// (the raise runs inside the save hook — codex P3).
471	639	$started = microtime( true );
472	640	$this->index->raise_group_concat_max_len();
473		- $result = $wpdb->query(
	641	+ $this->query_with_retry(
474	642	$wpdb->prepare(
475	643	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
476	644	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
477	645	. ' FROM (' . $this->index->row_digest_select_sql( 'p.ID = %d' ) . ') t'
		@@ -476,14 +644,12 @@
476	644	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
477	645	. ' FROM (' . $this->index->row_digest_select_sql( 'p.ID = %d' ) . ') t'
478	646	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
479	647	$post_id
480		- )
	648	+ ),
	649	+ 'upsert stored digest failed: ',
	650	+ $started
481	651	);
482		- self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
483		- if ( false === $result ) {
484		- throw new RuntimeException( 'upsert stored digest failed: ' . $wpdb->last_error );
485		- }
486	652	}
487	653
488	654	/**
489	655	* Customer analogue of {@see upsert_digest} (ADR 0015, Leg-3 phase 7):
		@@ -493,9 +659,9 @@
493	659	public function upsert_customer_digest( int $user_id ): void {
494	660	global $wpdb;
495	661	$started = microtime( true );
496	662	$this->index->raise_group_concat_max_len();
497		- $result = $wpdb->query(
	663	+ $this->query_with_retry(
498	664	$wpdb->prepare(
499	665	'INSERT INTO ' . $this->table_name() . ' (object_type, object_id, digest, updated_gmt)'
500	666	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
501	667	. ' FROM (' . $this->index->customer_digest_select_sql( 'u.ID = %d' ) . ') t'
		@@ -500,14 +666,64 @@
500	666	. ' SELECT t.object_type, t.id, t.crc, UTC_TIMESTAMP()'
501	667	. ' FROM (' . $this->index->customer_digest_select_sql( 'u.ID = %d' ) . ') t'
502	668	. ' ON DUPLICATE KEY UPDATE digest = VALUES(digest), updated_gmt = VALUES(updated_gmt)',
503	669	$user_id
504		- )
	670	+ ),
	671	+ 'upsert stored customer digest failed: ',
	672	+ $started
505	673	);
	674	+ }
	675	+
	676	+ /**
	677	+ * MySQL/MariaDB error numbers a second attempt can clear: 1020 ER_CHECKREAD
	678	+ * ("Record has changed since last read"), 1205 ER_LOCK_WAIT_TIMEOUT, 1213
	679	+ * ER_LOCK_DEADLOCK. Two requests upserting the same digest row race on
	680	+ * the `INSERT … ON DUPLICATE KEY UPDATE`; the retry reads the updated row.
	681	+ */
	682	+ private const TRANSIENT_CONTENTION_ERRNOS = array( 1020, 1205, 1213 );
	683	+
	684	+ /**
	685	+ * Message fallback for the same three errors, used only when the driver's
	686	+ * error number is unavailable (a wpdb without a live mysqli handle).
	687	+ */
	688	+ private const TRANSIENT_CONTENTION_MESSAGES = array(
	689	+ 'Record has changed since last read',
	690	+ 'Lock wait timeout',
	691	+ 'Deadlock found',
	692	+ );
	693	+
	694	+ /** Retry a contended upsert once, including both attempts in the hook timing. */
	695	+ private function query_with_retry( string $sql, string $error_message, float $started ): void {
	696	+ global $wpdb;
	697	+ $result = $wpdb->query( $sql );
	698	+ if ( false === $result && $this->is_transient_contention( $wpdb ) ) {
	699	+ $result = $wpdb->query( $sql );
	700	+ }
506	701	self::$request_write_ms += ( microtime( true ) - $started ) * 1000;
507	702	if ( false === $result ) {
508		- throw new RuntimeException( 'upsert stored customer digest failed: ' . $wpdb->last_error );
	703	+ throw new RuntimeException( $error_message . $wpdb->last_error );
509	704	}
	705	+ }
	706	+
	707	+ /**
	708	+ * The error number is authoritative: server messages are localised
	709	+ * (`lc_messages`), so the English text is only a fallback for a handle-less
	710	+ * wpdb. `$wpdb->dbh` is reachable through wpdb's magic getter.
	711	+ */
	712	+ private function is_transient_contention( \wpdb $wpdb ): bool {
	713	+ $dbh = $wpdb->__get( 'dbh' );
	714	+ if ( $dbh instanceof \mysqli ) {
	715	+ $errno = mysqli_errno( $dbh ); // phpcs:ignore WordPress.DB.RestrictedFunctions -- reads the driver's last error number; no query is issued.
	716	+ if ( 0 !== $errno ) {
	717	+ return in_array( $errno, self::TRANSIENT_CONTENTION_ERRNOS, true );
	718	+ }
	719	+ }
	720	+ foreach ( self::TRANSIENT_CONTENTION_MESSAGES as $message ) {
	721	+ if ( false !== strpos( $wpdb->last_error, $message ) ) {
	722	+ return true;
	723	+ }
	724	+ }
	725	+ return false;
510	726	}
511	727
512	728	/**
513	729	* Backfill/repair: prune orphans, then digest every live row in one