404-solution / includes / diagnostics / ActiveOperationBreadcrumbs.php

ActiveOperationBreadcrumbs.php in 404 Solution trunk, at includes/diagnostics/ActiveOperationBreadcrumbs.php

489 lines 19.7 KB

Raw Download Zip

1	<?php
2
3	if (!defined('ABSPATH')) {
4	exit;
5	}
6
7	/**
8	* Bounded append-only state for operations that run after verbose caps.
9	*
10	* One flushed JSONL record is appended for every transition. Readers fold the
11	* stream by (request_id, boundary), keeping the latest 32 live pairs. The raw
12	* journal is capped at 1 MiB; only the writer that would cross that bound pays
13	* for an atomic compaction to the folded table.
14	*
15	* A worker death during append can leave a final unterminated fragment. Readers
16	* discard only that final fragment and retain the complete prefix; the next
17	* writer compacts it away before appending. A malformed terminated line still
18	* fails closed. Compaction itself writes and flushes a complete snapshot before
19	* renaming, so a death leaves either the append-only source or the complete
20	* compacted file. No deliberately truncated target is exposed.
21	*
22	* This class owns the bounded persistence contract. Boundary field privacy is
23	* owned by ABJ_404_Solution_ActiveOperationBoundaryManifest, and event semantics
24	* remain with AjaxCheckpointLogger.
25	*
26	* @phpstan-type EncodedRecord array{record: array<string, mixed>, line: string}
27	* @phpstan-type Snapshot array{ino: mixed, dev: mixed, size: int,
28	* max_sequence: int, torn_tail: bool, records: array<int, EncodedRecord>}
29	*/
30	final class ABJ_404_Solution_ActiveOperationBreadcrumbs {
31
32	const FILE = 'abj404_ajax_active_operations.jsonl';
33	const LOCK_FILE = 'abj404_ajax_active_operations.lock';
34	const MAX_RECORDS = 32;
35	const MAX_RECORD_BYTES = 2048;
36	const MAX_FILE_BYTES = 1048576;
37	const LOCK_WAIT_TIMEOUT_US = 50000;
38
39	/**
40	* Last validated fold per path. Same-inode growth is parsed from the old
41	* size, so a sibling append joins the fold before this process compacts.
42	*
43	* @var array<string, Snapshot>
44	*/
45	private static $rememberedTable = array();
46
47	/**
48	* Append the latest transition for one request/boundary pair.
49	*
50	* @param array<string, mixed> $record
51	* @return array{status: string, reason: string}
52	*/
53	public static function replace(string $directory, array $record): array {
54	try {
55	$validated = self::validateRecord($record);
56	if ($validated['status'] !== 'complete') {
57	return $validated;
58	}
59	$record = self::sanitizeRecord($record);
60	if (!class_exists('ABJ_404_Solution_FileSystemService')
61	\|\| !ABJ_404_Solution_FileSystemService::createDirectoryWithErrorMessages($directory)) {
62	return self::failure('directory_unavailable');
63	}
64
65	$path = $directory . self::FILE;
66	$lockPath = $directory . self::LOCK_FILE;
67	$acquired = ABJ_404_Solution_DiagnosticAppendStream::acquireExclusive(
68	$lockPath,
69	self::LOCK_WAIT_TIMEOUT_US
70	);
71	if ($acquired['status'] === 'failed') {
72	self::reportFailure('active-operation lock could not be opened: ' . $lockPath);
73	return self::failure('lock_open_failed');
74	}
75	try {
76	if ($acquired['status'] === 'lock_timeout') {
77	self::reportFailure('active-operation lock wait exceeded: ' . $lockPath);
78	return self::failure('lock_wait_exceeded');
79	}
80	$snapshot = self::readExistingOrRemembered($path);
81	if ($snapshot === null) {
82	return self::failure('existing_file_unparseable');
83	}
84
85	$record['breadcrumb_format_version'] = 2;
86	$record['breadcrumb_seq'] = $snapshot['max_sequence'] >= PHP_INT_MAX
87	? 1 : $snapshot['max_sequence'] + 1;
88	$encoded = self::encodeRecord($record);
89	if ($encoded === null) {
90	self::reportFailure('active-operation file contains an over-budget record.');
91	return self::failure('record_too_large');
92	}
93	$payloadBytes = strlen($encoded['line']) + 1;
94	if ($snapshot['torn_tail']
95	\|\| $snapshot['size'] + $payloadBytes > self::MAX_FILE_BYTES) {
96	$snapshot = self::compactFile($path, $snapshot);
97	if ($snapshot === null) {
98	return self::failure('compaction_failed');
99	}
100	}
101
102	$append = ABJ_404_Solution_DiagnosticAppendStream::append(
103	$path,
104	$encoded['line'] . "\n"
105	);
106	if ($append['status'] !== 'complete') {
107	unset(self::$rememberedTable[$path]);
108	self::reportFailure('active-operation append failed: '
109	. ($append['reason'] ?? 'unknown'));
110	return self::failure('append_failed');
111	}
112	$snapshot['records'] = self::foldEncodedRecord($snapshot['records'], $encoded);
113	$snapshot['size'] += $payloadBytes;
114	$snapshot['max_sequence'] = $record['breadcrumb_seq'];
115	$snapshot['torn_tail'] = false;
116	self::rememberSnapshot($path, $snapshot);
117	return array('status' => 'complete', 'reason' => '');
118	} finally {
119	if (ABJ_404_Solution_DiagnosticAppendStream::release($lockPath)['status'] === 'failed') {
120	self::reportFailure('active-operation lock could not be released: ' . $lockPath);
121	}
122	}
123	} catch (Throwable $e) {
124	self::reportFailure('active-operation replacement failed: ' . $e->getMessage());
125	return self::failure('unexpected_failure');
126	}
127	}
128
129	/** The active-state file path whether or not the file currently exists. */
130	public static function path(string $directory): string {
131	return $directory . self::FILE;
132	}
133
134	/**
135	* Return the latest active identities for one ledger request.
136	*
137	* @return array<int, array<string, mixed>>
138	*/
139	public static function activeForRequest(string $directory, string $requestId): array {
140	if (preg_match('/^[A-Za-z0-9]{8,64}$/', $requestId) !== 1) {
141	return array();
142	}
143	$snapshot = self::readSnapshot(self::path($directory));
144	if ($snapshot === null) {
145	return array();
146	}
147	return array_values(array_filter(
148	array_column($snapshot['records'], 'record'),
149	static function (array $record) use ($requestId): bool {
150	return ($record['request_id'] ?? null) === $requestId
151	&& ($record['state'] ?? null) === 'active';
152	}
153	));
154	}
155
156	/**
157	* Fold active-operation lines for support consumers while preserving every
158	* unrelated or malformed line for its owning policy to inspect.
159	*
160	* @param array<int, string> $lines
161	* @return array<int, string>
162	*/
163	public static function compactSupportLines(array $lines): array {
164	$latest = array();
165	foreach ($lines as $index => $line) {
166	$record = json_decode($line, true);
167	$key = is_array($record) ? self::recordKey($record) : '';
168	if ($key === '') {
169	continue;
170	}
171	unset($latest[$key]);
172	$latest[$key] = $index;
173	if (count($latest) > self::MAX_RECORDS) {
174	array_shift($latest);
175	}
176	}
177	if ($latest === array()) {
178	return $lines;
179	}
180	$keep = array_fill_keys(array_values($latest), true);
181	return array_values(array_filter(
182	$lines,
183	static function (string $line, int $index) use ($keep): bool {
184	$record = json_decode($line, true);
185	return !is_array($record)
186	\|\| self::recordKey($record) === ''
187	\|\| isset($keep[$index]);
188	},
189	ARRAY_FILTER_USE_BOTH
190	));
191	}
192
193	/** Discard request-local cached state between PHPUnit request fixtures. */
194	public static function resetForTests(): void {
195	self::$rememberedTable = array();
196	}
197
198	/**
199	* @param array<string, mixed> $record
200	* @return array<string, mixed>
201	*/
202	private static function sanitizeRecord(array $record): array {
203	$boundary = is_string($record['boundary'] ?? null) ? $record['boundary'] : '';
204	$core = array();
205	foreach (array('request_id', 'event', 'boundary', 'state', 'checkpoint_id') as $field) {
206	if (array_key_exists($field, $record)) {
207	$core[$field] = $record[$field];
208	}
209	}
210	return array_merge(
211	$core,
212	ABJ_404_Solution_ActiveOperationBoundaryManifest::selectFields($boundary, $record)
213	);
214	}
215
216	/**
217	* @param array<mixed, mixed> $record
218	* @return array{status: string, reason: string}
219	*/
220	private static function validateRecord(array $record): array {
221	$requestId = $record['request_id'] ?? null;
222	$boundary = $record['boundary'] ?? null;
223	$state = $record['state'] ?? null;
224	if (!is_string($requestId) \|\| preg_match('/^[A-Za-z0-9]{8,64}$/', $requestId) !== 1) {
225	self::reportFailure('active-operation record has an invalid request id.');
226	return self::failure('invalid_request_id');
227	}
228	if (!is_string($boundary)
229	\|\| !ABJ_404_Solution_ActiveOperationBoundaryManifest::hasBoundary($boundary)) {
230	self::reportFailure('active-operation record has an invalid boundary.');
231	return self::failure('invalid_boundary');
232	}
233	if (!in_array($state, array('active', 'complete'), true)) {
234	self::reportFailure('active-operation record has an invalid state.');
235	return self::failure('invalid_state');
236	}
237	$encoded = json_encode($record, JSON_UNESCAPED_SLASHES);
238	if (!is_string($encoded) \|\| strlen($encoded) > self::MAX_RECORD_BYTES) {
239	self::reportFailure('active-operation record exceeds its fixed record budget.');
240	return self::failure('record_too_large');
241	}
242	return array('status' => 'complete', 'reason' => '');
243	}
244
245	/**
246	* @param array<string, mixed> $record
247	* @return array{record: array<string, mixed>, line: string}\|null
248	*/
249	private static function encodeRecord(array $record): ?array {
250	$line = json_encode($record, JSON_UNESCAPED_SLASHES);
251	if (!is_string($line) \|\| strlen($line) > self::MAX_RECORD_BYTES) {
252	return null;
253	}
254	return array('record' => $record, 'line' => $line);
255	}
256
257	/** @return Snapshot\|null */
258	private static function readExistingOrRemembered(string $path): ?array {
259	clearstatcache(true, $path);
260	$current = @stat($path);
261	if (!is_array($current) \|\| !@is_file($path)) {
262	return self::emptySnapshot();
263	}
264	$size = is_int($current['size'] ?? null) ? $current['size'] : -1;
265	if ($size < 0 \|\| $size > self::MAX_FILE_BYTES) {
266	self::reportFailure('active-operation file exceeds its fixed byte bound: ' . $path);
267	return null;
268	}
269	$remembered = self::$rememberedTable[$path] ?? null;
270	if (is_array($remembered)
271	&& $remembered['ino'] === ($current['ino'] ?? null)
272	&& $remembered['dev'] === ($current['dev'] ?? null)
273	&& $size >= $remembered['size']) {
274	if ($size === $remembered['size']) {
275	return $remembered;
276	}
277	return self::readSnapshot($path, $remembered['size'], $remembered);
278	}
279	return self::readSnapshot($path);
280	}
281
282	/**
283	* @param Snapshot\|null $base Previously validated prefix.
284	* @return Snapshot\|null
285	*/
286	private static function readSnapshot(string $path, int $offset = 0, ?array $base = null): ?array {
287	if (!@is_file($path)) {
288	return self::emptySnapshot();
289	}
290	$handle = @fopen($path, 'rb');
291	if ($handle === false) {
292	self::reportFailure('active-operation file could not be read: ' . $path);
293	return null;
294	}
295	try {
296	$stat = @fstat($handle);
297	$size = is_array($stat) && is_int($stat['size'] ?? null) ? $stat['size'] : -1;
298	if ($size < 0 \|\| $size > self::MAX_FILE_BYTES \|\| $offset > $size) {
299	self::reportFailure('active-operation file exceeds its fixed byte bound: ' . $path);
300	return null;
301	}
302	if ($offset > 0 && @fseek($handle, $offset) !== 0) {
303	self::reportFailure('active-operation file suffix could not be read: ' . $path);
304	return null;
305	}
306	$snapshot = $base ?? self::emptySnapshot();
307	$snapshot['ino'] = is_array($stat) ? ($stat['ino'] ?? null) : null;
308	$snapshot['dev'] = is_array($stat) ? ($stat['dev'] ?? null) : null;
309	$snapshot['size'] = $size;
310	$snapshot['torn_tail'] = false;
311	while (($line = @fgets($handle, self::MAX_RECORD_BYTES + 3)) !== false) {
312	$parsed = self::parseJournalLine($line, @feof($handle), $path);
313	if ($parsed['status'] === 'torn') {
314	$snapshot['torn_tail'] = true;
315	break;
316	}
317	if ($parsed['status'] !== 'complete') {
318	return null;
319	}
320	$snapshot['records'] = self::foldEncodedRecord(
321	$snapshot['records'],
322	$parsed['encoded']
323	);
324	$snapshot['max_sequence'] = max(
325	$snapshot['max_sequence'],
326	$parsed['sequence']
327	);
328	}
329	return $snapshot;
330	} finally {
331	@fclose($handle);
332	}
333	}
334
335	/** @return Snapshot */
336	private static function emptySnapshot(): array {
337	return array(
338	'ino' => null,
339	'dev' => null,
340	'size' => 0,
341	'max_sequence' => 0,
342	'torn_tail' => false,
343	'records' => array(),
344	);
345	}
346
347	/**
348	* Parse one bounded journal read without letting a torn final append poison
349	* the complete prefix.
350	*
351	* @return array{status: 'complete', encoded: EncodedRecord, sequence: int}
352	* \|array{status: 'torn'\|'failed'}
353	*/
354	private static function parseJournalLine(string $line, bool $atEof, string $path): array {
355	if (substr($line, -1) !== "\n") {
356	if ($atEof) {
357	return array('status' => 'torn');
358	}
359	self::reportFailure('active-operation file contains an over-budget record: ' . $path);
360	return array('status' => 'failed');
361	}
362	$line = rtrim($line, "\r\n");
363	if (strlen($line) > self::MAX_RECORD_BYTES) {
364	self::reportFailure('active-operation file contains an over-budget record: ' . $path);
365	return array('status' => 'failed');
366	}
367	$decoded = json_decode($line, true);
368	if (!is_array($decoded) \|\| self::recordKey($decoded) === '') {
369	self::reportFailure('active-operation file contains an unparseable record: ' . $path);
370	return array('status' => 'failed');
371	}
372	$record = array();
373	foreach ($decoded as $key => $value) {
374	$record[(string)$key] = $value;
375	}
376	$sequence = is_int($record['breadcrumb_seq'] ?? null)
377	? $record['breadcrumb_seq'] : 0;
378	return array(
379	'status' => 'complete',
380	'encoded' => array('record' => $record, 'line' => $line),
381	'sequence' => $sequence,
382	);
383	}
384
385	/**
386	* @param array<int, array{record: array<string, mixed>, line: string}> $records
387	* @param array{record: array<string, mixed>, line: string} $replacement
388	* @return array<int, array{record: array<string, mixed>, line: string}>
389	*/
390	private static function foldEncodedRecord(array $records, array $replacement): array {
391	$key = self::recordKey($replacement['record']);
392	$kept = array_values(array_filter(
393	$records,
394	static function (array $encoded) use ($key): bool {
395	return self::recordKey($encoded['record']) !== $key;
396	}
397	));
398	$kept[] = $replacement;
399	return count($kept) > self::MAX_RECORDS
400	? array_slice($kept, -self::MAX_RECORDS)
401	: $kept;
402	}
403
404	/** @param array<mixed, mixed> $record */
405	private static function recordKey(array $record): string {
406	$requestId = $record['request_id'] ?? null;
407	$boundary = $record['boundary'] ?? null;
408	$state = $record['state'] ?? null;
409	if (($record['event'] ?? '') !== 'active_operation_breadcrumb'
410	\|\| !is_string($requestId)
411	\|\| preg_match('/^[A-Za-z0-9]{8,64}$/', $requestId) !== 1
412	\|\| !is_string($boundary)
413	\|\| !ABJ_404_Solution_ActiveOperationBoundaryManifest::hasBoundary($boundary)
414	\|\| !in_array($state, array('active', 'complete'), true)) {
415	return '';
416	}
417	return $requestId . '\|' . $boundary;
418	}
419
420	/**
421	* @param Snapshot $snapshot Validated folded state.
422	* @return Snapshot\|null Compacted snapshot, or null on failure.
423	*/
424	private static function compactFile(string $path, array $snapshot): ?array {
425	$lines = array_column($snapshot['records'], 'line');
426	$payload = implode("\n", $lines) . ($lines === array() ? '' : "\n");
427	$temporary = $path . '.compact.tmp';
428	$handle = @fopen($temporary, 'wb');
429	if ($handle === false) {
430	self::reportFailure('active-operation compaction file could not be opened: ' . $temporary);
431	return null;
432	}
433	try {
434	$written = @fwrite($handle, $payload);
435	$flushed = @fflush($handle);
436	} finally {
437	@fclose($handle);
438	}
439	if ($written !== strlen($payload) \|\| !$flushed) {
440	self::reportFailure('active-operation compaction file could not be flushed: '
441	. $temporary);
442	@unlink($temporary);
443	return null;
444	}
445	if (!@rename($temporary, $path)) {
446	self::reportFailure('active-operation compacted file could not be atomically replaced: '
447	. $path);
448	@unlink($temporary);
449	return null;
450	}
451	ABJ_404_Solution_DiagnosticAppendStream::invalidate($path);
452	clearstatcache(true, $path);
453	$stat = @stat($path);
454	if (!is_array($stat)) {
455	self::reportFailure('active-operation compacted file identity could not be read: ' . $path);
456	return null;
457	}
458	$snapshot['ino'] = $stat['ino'] ?? null;
459	$snapshot['dev'] = $stat['dev'] ?? null;
460	$snapshot['size'] = strlen($payload);
461	$snapshot['torn_tail'] = false;
462	return $snapshot;
463	}
464
465	/** @param Snapshot $snapshot Validated folded state. */
466	private static function rememberSnapshot(string $path, array $snapshot): void {
467	if ($snapshot['ino'] === null \|\| $snapshot['dev'] === null) {
468	clearstatcache(true, $path);
469	$stat = @stat($path);
470	if (!is_array($stat)) {
471	unset(self::$rememberedTable[$path]);
472	return;
473	}
474	$snapshot['ino'] = $stat['ino'] ?? null;
475	$snapshot['dev'] = $stat['dev'] ?? null;
476	}
477	self::$rememberedTable[$path] = $snapshot;
478	}
479
480	/** @return array{status: string, reason: string} */
481	private static function failure(string $reason): array {
482	return array('status' => 'failed', 'reason' => $reason);
483	}
484
485	private static function reportFailure(string $message): void {
486	abj404_logPhpFallback('active-operation-breadcrumb', $message);
487	}
488	}
489