404-solution / includes / logs / LogsQueueFlusher.php

LogsQueueFlusher.php in 404 Solution 4.3.0, at includes/logs/LogsQueueFlusher.php

287 lines 11.7 KB

Raw Download Zip

1	<?php
2
3	if (!defined('ABSPATH')) {
4	exit;
5	}
6
7	/**
8	* Owns deferred log queue flushing, including batch INSERT construction,
9	* schema-column validation, and per-row fallback when a batch fails.
10	*/
11	class ABJ_404_Solution_LogsQueueFlusher {
12
13	/** @var ABJ_404_Solution_DatabaseCore */
14	private $dbCore;
15
16	/** @var ABJ_404_Solution_Logging */
17	private $logger;
18
19	/** @var ABJ_404_Solution_LogsEntrySanitizer */
20	private $entrySanitizer;
21
22	/** @var ABJ_404_Solution_LogsWriteRecoveryPolicy */
23	private $recovery;
24
25	/**
26	* @param ABJ_404_Solution_DatabaseCore $dbCore
27	* @param ABJ_404_Solution_Logging $logger
28	* @param ABJ_404_Solution_LogsEntrySanitizer $entrySanitizer
29	* @param ABJ_404_Solution_LogsWriteRecoveryPolicy $recovery
30	*/
31	public function __construct(
32	ABJ_404_Solution_DatabaseCore $dbCore,
33	$logger,
34	ABJ_404_Solution_LogsEntrySanitizer $entrySanitizer,
35	ABJ_404_Solution_LogsWriteRecoveryPolicy $recovery
36	) {
37	$this->dbCore = $dbCore;
38	$this->logger = $logger;
39	$this->entrySanitizer = $entrySanitizer;
40	$this->recovery = $recovery;
41	}
42
43	/**
44	* @param array<int, array<string, mixed>> $queue
45	* @param-out array<int, array<string, mixed>> $queue
46	* @param callable $flushCallback
47	* @param array<string, mixed> $entry
48	*/
49	public function queueLogEntry(array &$queue, bool &$shutdownHookRegistered, callable $flushCallback, array $entry): void {
50	$queue[] = $entry;
51	if (!$shutdownHookRegistered) {
52	$shutdownHookRegistered = true;
53	add_action('shutdown', $flushCallback, 9);
54	}
55	}
56
57	/**
58	* @param array<int, array<string, mixed>> $queue
59	*/
60	public function flushLogQueue(array &$queue, bool &$shutdownHookRegistered, bool &$isFlushingLogQueue): void {
61	if ($isFlushingLogQueue) {
62	return;
63	}
64	$isFlushingLogQueue = true;
65
66	try {
67	if (empty($queue)) {
68	return;
69	}
70
71	global $wpdb;
72	$tableName = $this->dbCore->doTableNameReplacements('{wp_abj404_logsv2}');
73
74	$columns = array_keys($queue[0]);
75	$validatedColumns = [];
76	foreach ($columns as $col) {
77	if (preg_match('/^[a-z_][a-z0-9_]*$/i', $col)) {
78	$validatedColumns[] = $col;
79	}
80	}
81	$schemaColumns = $this->dbCore->tableNameResolver()->getTableColumnNames($tableName);
82	if (!empty($schemaColumns)) {
83	$validatedColumns = array_intersect($validatedColumns, $schemaColumns);
84	}
85	if (empty($validatedColumns)) {
86	return;
87	}
88
89	$columnList = '`' . implode('`, `', $validatedColumns) . '`';
90	$sanitizedEntries = [];
91	foreach ($queue as $entry) {
92	$entryColumns = array_keys($entry);
93	$missingCols = array_diff($validatedColumns, $entryColumns);
94	if (!empty($missingCols)) {
95	continue;
96	}
97	$sanitized = $this->entrySanitizer->sanitizeLogEntry($entry);
98	if ($sanitized === null) {
99	continue;
100	}
101	$sanitizedEntries[] = $sanitized;
102	}
103	if (empty($sanitizedEntries)) {
104	return;
105	}
106
107	list($formats, $flattenedValues) = $this->buildValuePlaceholders($sanitizedEntries, $validatedColumns);
108	$sql = "INSERT IGNORE INTO `{$tableName}` ({$columnList}) VALUES " . implode(', ', $formats);
109	// DAO-bypass-approved: queue flusher batches dynamic INSERT placeholders and must inspect wpdb last_error on the same connection.
110	$prepared = $wpdb->prepare($sql, $flattenedValues);
111	$wpdb->flush();
112	// DAO-bypass-approved: batch log insert must preserve same-handle last_error for recovery classification.
113	$result = $wpdb->query($prepared);
114
115	if ($result === false && !empty($wpdb->last_error)) {
116	$this->recoverFailedBatch($tableName, $columnList, $prepared, $sql, $flattenedValues, $sanitizedEntries, $validatedColumns, (string)$wpdb->last_error);
117	}
118	} finally {
119	$queue = [];
120	$shutdownHookRegistered = false;
121	$isFlushingLogQueue = false;
122	}
123	}
124
125	/**
126	* @param array<int, array<string, mixed>> $entries
127	* @param array<int, string> $validatedColumns
128	* @return array{0: array<int, string>, 1: array<int, mixed>}
129	*/
130	private function buildValuePlaceholders(array $entries, array $validatedColumns): array {
131	$formats = [];
132	$flattenedValues = [];
133	foreach ($entries as $entry) {
134	$rowFormats = [];
135	foreach ($validatedColumns as $col) {
136	$value = $entry[$col];
137	if ($value === null) {
138	$rowFormats[] = 'NULL';
139	continue;
140	}
141	if (is_int($value)) {
142	$rowFormats[] = '%d';
143	} else {
144	$rowFormats[] = '%s';
145	}
146	$flattenedValues[] = $value;
147	}
148	$formats[] = '(' . implode(', ', $rowFormats) . ')';
149	}
150	return array($formats, $flattenedValues);
151	}
152
153	/**
154	* @param array<int, mixed> $flattenedValues
155	* @param array<int, array<string, mixed>> $sanitizedEntries
156	* @param array<int, string> $validatedColumns
157	*/
158	private function recoverFailedBatch(
159	string $tableName,
160	string $columnList,
161	string $prepared,
162	string $sql,
163	array $flattenedValues,
164	array $sanitizedEntries,
165	array $validatedColumns,
166	string $batchError
167	): void {
168	global $wpdb;
169
170	if ($this->recovery->isTableFullError($batchError)) {
171	$trimmed = $this->recovery->autoTrimLogsv2IfNeeded($tableName, $batchError);
172	if ($trimmed) {
173	$wpdb->flush();
174	// DAO-bypass-approved: table-full recovery retries the already-prepared batch on the same wpdb connection.
175	$retryResult = $wpdb->query($prepared);
176	if ($retryResult !== false) {
177	return;
178	}
179	$batchError = (string)$wpdb->last_error;
180	}
181	$this->recovery->setLogsv2FullNotice($batchError);
182	}
183
184	if ($this->recovery->isCommandsOutOfSyncError($batchError)) {
185	$isolated = $this->recovery->getIsolatedWpdb();
186	if ($isolated !== null) {
187	$isolated->flush();
188	$isolatedResult = $isolated->query($prepared);
189	if ($isolatedResult !== false) {
190	$context = $this->recovery->getWpdbRecentQueryContextForLogs();
191	$suffix = ($context !== '') ? " \| savequeries_context={$context}" : '';
192	$this->logger->warn("flushLogQueue batch INSERT succeeded using isolated DB connection (commands out of sync on shared connection).{$suffix}");
193	return;
194	}
195	$batchError .= " \| isolated_error=" . $isolated->last_error;
196	} else {
197	$batchError .= " \| isolated_error=no_isolated_connection";
198	}
199	}
200
201	$this->recoverEntriesIndividually($tableName, $columnList, $sanitizedEntries, $validatedColumns, $batchError);
202	}
203
204	/**
205	* @param array<int, array<string, mixed>> $sanitizedEntries
206	* @param array<int, string> $validatedColumns
207	*/
208	private function recoverEntriesIndividually(
209	string $tableName,
210	string $columnList,
211	array $sanitizedEntries,
212	array $validatedColumns,
213	string $batchError
214	): void {
215	global $wpdb;
216	$successCount = 0;
217	$failCount = 0;
218	$failureDetails = [];
219
220	foreach ($sanitizedEntries as $index => $entry) {
221	$rowFormats = [];
222	$rowValues = [];
223	foreach ($validatedColumns as $col) {
224	$value = $entry[$col];
225	if ($value === null) {
226	$rowFormats[] = 'NULL';
227	} else {
228	$rowFormats[] = is_int($value) ? '%d' : '%s';
229	$rowValues[] = $value;
230	}
231	}
232	$rowPlaceholder = '(' . implode(', ', $rowFormats) . ')';
233	/** @var literal-string $singleSqlTemplate */
234	$singleSqlTemplate = "INSERT IGNORE INTO `{$tableName}` ({$columnList}) VALUES {$rowPlaceholder}";
235	// DAO-bypass-approved: row-level fallback reuses wpdb prepare/query so recovery can classify last_error per row.
236	$singleSql = $wpdb->prepare($singleSqlTemplate, $rowValues);
237	$wpdb->flush();
238	// DAO-bypass-approved: row-level retry must preserve same-handle last_error for per-row recovery.
239	$singleResult = $wpdb->query((string)$singleSql);
240
241	if ($singleResult === false && !empty($wpdb->last_error)) {
242	$lastError = (string)$wpdb->last_error;
243	if ($this->recovery->isCommandsOutOfSyncError($wpdb->last_error)) {
244	$isolated = $this->recovery->getIsolatedWpdb();
245	if ($isolated !== null) {
246	$isolated->flush();
247	$isolatedSingleSql = $isolated->prepare($singleSqlTemplate, $rowValues);
248	$isolatedSingleResult = $isolated->query((string)$isolatedSingleSql);
249	if ($isolatedSingleResult !== false) {
250	$successCount++;
251	continue;
252	}
253	$lastError = $lastError . " \| isolated_error=" . $isolated->last_error;
254	} else {
255	$lastError = $lastError . " \| isolated_error=no_isolated_connection";
256	}
257	}
258	$failCount++;
259	$payload = function_exists('wp_json_encode') ? wp_json_encode($entry) : json_encode($entry);
260	if (is_string($payload) && strlen($payload) > 1024) {
261	$payload = substr($payload, 0, 1024) . '...';
262	}
263	$failureDetails[] = ['index' => $index, 'error' => $lastError, 'payload' => $payload];
264	} else {
265	$successCount++;
266	}
267	}
268
269	if ($failCount > 0) {
270	$detailsParts = [];
271	foreach (array_slice($failureDetails, 0, 3) as $detail) {
272	$detailsParts[] = "entry {$detail['index']}: {$detail['error']} \| payload={$detail['payload']}";
273	}
274	$detailsSuffix = count($failureDetails) > 3 ? ' \| (additional failures omitted)' : '';
275	$context = $this->recovery->getWpdbRecentQueryContextForLogs();
276	$contextSuffix = ($context !== '') ? (" \| savequeries_context=" . $context) : '';
277	if ($this->dbCore->errorClassifier()->classifyAndHandleInfrastructureError($batchError)) {
278	$this->logger->warn("flushLogQueue recovery incomplete: {$successCount} inserted, {$failCount} failed. \| batch_error=" . $batchError . " \| failures=" . implode(' \|\| ', $detailsParts) . $detailsSuffix . $contextSuffix);
279	} else {
280	$this->logger->errorMessage("flushLogQueue recovery incomplete: {$successCount} inserted, {$failCount} failed. \| batch_error=" . $batchError . " \| failures=" . implode(' \|\| ', $detailsParts) . $detailsSuffix . $contextSuffix);
281	}
282	} else {
283	$this->logger->warn("flushLogQueue batch INSERT failed but recovered: all {$successCount} entries inserted individually. \| batch_error=" . $batchError);
284	}
285	}
286	}
287