404-solution / includes / logs / LogsQueueFlusher.php

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

289 lines 12.0 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	$batchDatabaseError = '';
66
67	try {
68	if (empty($queue)) {
69	return;
70	}
71
72	global $wpdb;
73	$tableName = $this->dbCore->doTableNameReplacements('{wp_abj404_logsv2}');
74
75	$columns = array_keys($queue[0]);
76	$validatedColumns = [];
77	foreach ($columns as $col) {
78	if (preg_match('/^[a-z_][a-z0-9_]*$/i', $col)) {
79	$validatedColumns[] = $col;
80	}
81	}
82	$schemaColumns = $this->dbCore->tableNameResolver()->getTableColumnNames($tableName);
83	if (!empty($schemaColumns)) {
84	$validatedColumns = array_intersect($validatedColumns, $schemaColumns);
85	}
86	if (empty($validatedColumns)) {
87	return;
88	}
89
90	$columnList = '`' . implode('`, `', $validatedColumns) . '`';
91	$sanitizedEntries = [];
92	foreach ($queue as $entry) {
93	$entryColumns = array_keys($entry);
94	$missingCols = array_diff($validatedColumns, $entryColumns);
95	if (!empty($missingCols)) {
96	continue;
97	}
98	$sanitized = $this->entrySanitizer->sanitizeLogEntry($entry);
99	if ($sanitized === null) {
100	continue;
101	}
102	$sanitizedEntries[] = $sanitized;
103	}
104	if (empty($sanitizedEntries)) {
105	return;
106	}
107
108	list($formats, $flattenedValues) = $this->buildValuePlaceholders($sanitizedEntries, $validatedColumns);
109	$sql = "INSERT IGNORE INTO `{$tableName}` ({$columnList}) VALUES " . implode(', ', $formats);
110	// DAO-bypass-approved: queue flusher batches dynamic INSERT placeholders and must inspect wpdb last_error on the same connection.
111	$prepared = $wpdb->prepare($sql, $flattenedValues);
112	$this->dbCore->connectionManager()->resetForRetry();
113	// DAO-bypass-approved: batch log insert must preserve same-handle last_error for recovery classification.
114	$result = $wpdb->query($prepared);
115
116	if ($result === false && !empty($wpdb->last_error)) {
117	$batchDatabaseError = (string)$wpdb->last_error;
118	$this->recoverFailedBatch($tableName, $columnList, $prepared, $sanitizedEntries, $validatedColumns, $batchDatabaseError);
119	}
120	} catch (\Throwable $e) {
121	// This runs on the 'shutdown' action, which fires on every
122	// request that queued a log entry. finally (below) already
123	// resets the queue/flags; without this catch, an uncaught
124	// throwable here would propagate out of do_action('shutdown')
125	// and abort any other plugin's later shutdown cleanup too, not
126	// just lose this batch of log rows.
127	$failureMessage = 'flushLogQueue failed: ' . get_class($e) . ': ' . $e->getMessage();
128	if ($batchDatabaseError !== '' && $this->dbCore->errorClassifier()->classifyAndHandleInfrastructureError($batchDatabaseError)) {
129	$this->logger->warn($failureMessage . ' \| database_error=' . $batchDatabaseError);
130	} else {
131	$this->logger->errorMessage(
132	$failureMessage,
133	$e instanceof \Exception ? $e : null
134	);
135	}
136	} finally {
137	$queue = [];
138	$shutdownHookRegistered = false;
139	$isFlushingLogQueue = false;
140	}
141	}
142
143	/**
144	* @param array<int, array<string, mixed>> $entries
145	* @param array<int, string> $validatedColumns
146	* @return array{0: array<int, string>, 1: array<int, mixed>}
147	*/
148	private function buildValuePlaceholders(array $entries, array $validatedColumns): array {
149	$formats = [];
150	$flattenedValues = [];
151	foreach ($entries as $entry) {
152	$rowFormats = [];
153	foreach ($validatedColumns as $col) {
154	$value = $entry[$col];
155	if ($value === null) {
156	$rowFormats[] = 'NULL';
157	continue;
158	}
159	if (is_int($value)) {
160	$rowFormats[] = '%d';
161	} else {
162	$rowFormats[] = '%s';
163	}
164	$flattenedValues[] = $value;
165	}
166	$formats[] = '(' . implode(', ', $rowFormats) . ')';
167	}
168	return array($formats, $flattenedValues);
169	}
170
171	/**
172	* @param array<int, array<string, mixed>> $sanitizedEntries
173	* @param array<int, string> $validatedColumns
174	*/
175	private function recoverFailedBatch(
176	string $tableName,
177	string $columnList,
178	string $prepared,
179	array $sanitizedEntries,
180	array $validatedColumns,
181	string $batchError
182	): void {
183	global $wpdb;
184
185	if ($this->recovery->isTableFullError($batchError)) {
186	$trimmed = $this->recovery->autoTrimLogsv2IfNeeded($tableName, $batchError);
187	if ($trimmed) {
188	if ($this->dbCore->connectionManager()->resetForRetry($batchError)) {
189	// DAO-bypass-approved: table-full recovery retries the already-prepared batch on the same wpdb connection.
190	$retryResult = $wpdb->query($prepared);
191	if ($retryResult !== false) {
192	return;
193	}
194	$batchError = (string)$wpdb->last_error;
195	}
196	}
197	$this->recovery->setLogsv2FullNotice($batchError);
198	}
199
200	$commandsOutOfSync = $this->dbCore->errorClassifier()
201	->taxonomy()
202	->connectivity()
203	->isCommandsOutOfSyncError($batchError);
204	if ($commandsOutOfSync && $this->dbCore->connectionManager()->resetForRetry($batchError)) {
205	// DAO-bypass-approved: centralized connection recovery drained error 2014 before this retry.
206	$retryResult = $wpdb->query($prepared);
207	if ($retryResult !== false) {
208	$context = $this->recovery->getWpdbRecentQueryContextForLogs();
209	$suffix = ($context !== '') ? " \| savequeries_context={$context}" : '';
210	$this->logger->warn("flushLogQueue batch INSERT recovered on the shared DB connection.{$suffix}");
211	return;
212	}
213	$batchError = (string)$wpdb->last_error;
214	}
215
216	$this->recoverEntriesIndividually($tableName, $columnList, $sanitizedEntries, $validatedColumns, $batchError);
217	}
218
219	/**
220	* @param array<int, array<string, mixed>> $sanitizedEntries
221	* @param array<int, string> $validatedColumns
222	*/
223	private function recoverEntriesIndividually(
224	string $tableName,
225	string $columnList,
226	array $sanitizedEntries,
227	array $validatedColumns,
228	string $batchError
229	): void {
230	global $wpdb;
231	$successCount = 0;
232	$failCount = 0;
233	$failureDetails = [];
234
235	foreach ($sanitizedEntries as $index => $entry) {
236	$rowFormats = [];
237	$rowValues = [];
238	foreach ($validatedColumns as $col) {
239	$value = $entry[$col];
240	if ($value === null) {
241	$rowFormats[] = 'NULL';
242	} else {
243	$rowFormats[] = is_int($value) ? '%d' : '%s';
244	$rowValues[] = $value;
245	}
246	}
247	$rowPlaceholder = '(' . implode(', ', $rowFormats) . ')';
248	/** @var literal-string $singleSqlTemplate */
249	$singleSqlTemplate = "INSERT IGNORE INTO `{$tableName}` ({$columnList}) VALUES {$rowPlaceholder}";
250	// DAO-bypass-approved: row-level fallback reuses wpdb prepare/query so recovery can classify last_error per row.
251	$singleSql = $wpdb->prepare($singleSqlTemplate, $rowValues);
252	$connectionReady = $this->dbCore->connectionManager()->resetForRetry($batchError);
253	// DAO-bypass-approved: row-level retry must preserve same-handle last_error for per-row recovery.
254	$singleResult = $connectionReady ? $wpdb->query((string)$singleSql) : false;
255
256	if ($singleResult === false) {
257	$lastError = !empty($wpdb->last_error)
258	? (string)$wpdb->last_error
259	: 'Connection reset failed before the row-level retry could execute.';
260	$failCount++;
261	$payload = function_exists('wp_json_encode') ? wp_json_encode($entry) : json_encode($entry);
262	if (is_string($payload) && strlen($payload) > 1024) {
263	$payload = substr($payload, 0, 1024) . '...';
264	}
265	$failureDetails[] = ['index' => $index, 'error' => $lastError, 'payload' => $payload];
266	} else {
267	$successCount++;
268	}
269	}
270
271	if ($failCount > 0) {
272	$detailsParts = [];
273	foreach (array_slice($failureDetails, 0, 3) as $detail) {
274	$detailsParts[] = "entry {$detail['index']}: {$detail['error']} \| payload={$detail['payload']}";
275	}
276	$detailsSuffix = count($failureDetails) > 3 ? ' \| (additional failures omitted)' : '';
277	$context = $this->recovery->getWpdbRecentQueryContextForLogs();
278	$contextSuffix = ($context !== '') ? (" \| savequeries_context=" . $context) : '';
279	if ($this->dbCore->errorClassifier()->classifyAndHandleInfrastructureError($batchError)) {
280	$this->logger->warn("flushLogQueue recovery incomplete: {$successCount} inserted, {$failCount} failed. \| batch_error=" . $batchError . " \| failures=" . implode(' \|\| ', $detailsParts) . $detailsSuffix . $contextSuffix);
281	} else {
282	$this->logger->errorMessage("flushLogQueue recovery incomplete: {$successCount} inserted, {$failCount} failed. \| batch_error=" . $batchError . " \| failures=" . implode(' \|\| ', $detailsParts) . $detailsSuffix . $contextSuffix);
283	}
284	} else {
285	$this->logger->warn("flushLogQueue batch INSERT failed but recovered: all {$successCount} entries inserted individually. \| batch_error=" . $batchError);
286	}
287	}
288	}
289