From c7641d7b4e4e46e211cfbe0edd674f6187677018 Mon Sep 17 00:00:00 2001 From: DariusIII Date: Wed, 29 Apr 2026 09:22:43 +0200 Subject: [PATCH] Constraint binaries to prevent OOM issues --- app/Services/Binaries/BinariesConfig.php | 15 ++- app/Services/Binaries/BinaryHandler.php | 126 +++++++++++------- app/Services/Binaries/CollectionHandler.php | 98 +++++++++----- .../Binaries/HeaderStorageService.php | 16 ++- .../Binaries/HeaderStorageTransaction.php | 53 ++++---- app/Services/Binaries/PartHandler.php | 78 +++++++---- .../Feature/BinariesStorageInternalsTest.php | 2 +- 7 files changed, 249 insertions(+), 139 deletions(-) diff --git a/app/Services/Binaries/BinariesConfig.php b/app/Services/Binaries/BinariesConfig.php index a1c3f136e..b7c6548a6 100644 --- a/app/Services/Binaries/BinariesConfig.php +++ b/app/Services/Binaries/BinariesConfig.php @@ -24,6 +24,17 @@ final readonly class BinariesConfig public int $partsChunkSize = 5000, public int $binariesUpdateChunkSize = 1000, public bool $echoCli = false, + // Number of headers processed (and bulk-inserted) at a time inside + // HeaderStorageService. This MUST stay small because each chunk + // produces multi-row INSERTs/SELECTs whose binding count and SQL size + // grow linearly with the value. Defaulting it to partsChunkSize + // (which used to only control single-row part flushes) caused MySQL + // and PHP to allocate hundreds of MB per scan and run out of RAM. + public int $headerChunkSize = 500, + // Hard upper bound applied internally to bulk SELECT/INSERT/UPDATE + // operations regardless of caller-provided chunk size, so a + // misconfiguration cannot blow up server memory. + public int $bulkSqlChunkSize = 500, ) {} /** @@ -41,8 +52,10 @@ final readonly class BinariesConfig partRepairLimit: self::getSettingInt('maxpartrepair', 15000), partRepairMaxTries: self::getSettingInt('partrepairmaxtries', 3), partsChunkSize: max(100, (int) config('nntmux.parts_chunk_size', 5000)), - binariesUpdateChunkSize: max(100, (int) config('nntmux.binaries_update_chunk_size', 1000)), + binariesUpdateChunkSize: max(100, min(1000, (int) config('nntmux.binaries_update_chunk_size', 1000))), echoCli: (bool) config('nntmux.echocli'), + headerChunkSize: max(50, min(2000, (int) config('nntmux.header_chunk_size', 500))), + bulkSqlChunkSize: max(50, min(1000, (int) config('nntmux.bulk_sql_chunk_size', 500))), ); } diff --git a/app/Services/Binaries/BinaryHandler.php b/app/Services/Binaries/BinaryHandler.php index 641679a31..949c09263 100644 --- a/app/Services/Binaries/BinaryHandler.php +++ b/app/Services/Binaries/BinaryHandler.php @@ -13,6 +13,14 @@ use Illuminate\Support\Facades\Log; */ final class BinaryHandler { + /** + * Hard upper bound on the number of rows packed into a single SQL + * statement (multi-row INSERT, OR-clause SELECT, etc.). Keeps the + * generated SQL string and PDO parameter list bounded regardless of + * how many headers the caller passes in. + */ + private const MAX_SQL_ROWS_PER_STATEMENT = 500; + /** @var array Pending binary updates */ private array $binariesUpdate = []; @@ -220,46 +228,50 @@ final class BinaryHandler /** @param list> $rows */ private function bulkInsertBinariesSqlite(array $rows): void { - $insertRows = []; - foreach ($rows as $row) { - $insertRows[] = [ - 'binaryhash' => $row['hash'], - 'name' => $row['name'], - 'collections_id' => $row['collections_id'], - 'totalparts' => $row['totalparts'], - 'currentparts' => 1, - 'filenumber' => $row['filenumber'], - 'partsize' => $row['partsize'], - ]; - } + foreach (array_chunk($rows, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $insertRows = []; + foreach ($chunk as $row) { + $insertRows[] = [ + 'binaryhash' => $row['hash'], + 'name' => $row['name'], + 'collections_id' => $row['collections_id'], + 'totalparts' => $row['totalparts'], + 'currentparts' => 1, + 'filenumber' => $row['filenumber'], + 'partsize' => $row['partsize'], + ]; + } - DB::table('binaries')->insertOrIgnore($insertRows); + DB::table('binaries')->insertOrIgnore($insertRows); + } } /** @param list> $rows */ private function bulkInsertBinariesMysql(array $rows): void { - $placeholders = []; - $bindings = []; - foreach ($rows as $row) { - $placeholders[] = '(UNHEX(?), ?, ?, ?, 1, ?, ?)'; - array_push( - $bindings, - $row['hash'], - $row['name'], - $row['collections_id'], - $row['totalparts'], - $row['filenumber'], - $row['partsize'] + foreach (array_chunk($rows, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = []; + $bindings = []; + foreach ($chunk as $row) { + $placeholders[] = '(UNHEX(?), ?, ?, ?, 1, ?, ?)'; + array_push( + $bindings, + $row['hash'], + $row['name'], + $row['collections_id'], + $row['totalparts'], + $row['filenumber'], + $row['partsize'] + ); + } + + DB::statement( + 'INSERT INTO binaries (binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) VALUES ' + .implode(',', $placeholders) + .' ON DUPLICATE KEY UPDATE currentparts = currentparts + 1, partsize = partsize + VALUES(partsize)', + $bindings ); } - - DB::statement( - 'INSERT INTO binaries (binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) VALUES ' - .implode(',', $placeholders) - .' ON DUPLICATE KEY UPDATE currentparts = currentparts + 1, partsize = partsize + VALUES(partsize)', - $bindings - ); } /** @@ -287,22 +299,39 @@ final class BinaryHandler } $driver = DB::getDriverName(); - $clauses = []; - $bindings = []; - foreach ($rows as $row) { - $clauses[] = $driver === 'sqlite' - ? '(binaryhash = ? AND collections_id = ?)' - : '(binaryhash = UNHEX(?) AND collections_id = ?)'; - $bindings[] = $row['hash']; - $bindings[] = $row['collections_id']; - } - $hashExpression = $driver === 'sqlite' ? 'binaryhash' : 'LOWER(HEX(binaryhash))'; - return DB::select( - "SELECT id, {$hashExpression} AS hashvalue, collections_id FROM binaries WHERE ".implode(' OR ', $clauses), - $bindings - ); + // Group lookups by collections_id and run small `binaryhash IN (...)` + // queries per collection. This avoids producing a single + // `WHERE (... AND ...) OR (... AND ...) OR ...` expression with + // thousands of clauses and bindings, which both bloats the SQL + // string in PHP and can force MySQL into pathological plans. + $rowsByCollection = []; + foreach ($rows as $row) { + $rowsByCollection[(int) $row['collections_id']][] = (string) $row['hash']; + } + + $results = []; + foreach ($rowsByCollection as $collectionId => $hashes) { + $hashes = array_values(array_unique($hashes)); + foreach (array_chunk($hashes, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), $driver === 'sqlite' ? '?' : 'UNHEX(?)')); + $bindings = $chunk; + $bindings[] = $collectionId; + + $rowsResult = DB::select( + "SELECT id, {$hashExpression} AS hashvalue, collections_id FROM binaries " + ."WHERE binaryhash IN ({$placeholders}) AND collections_id = ?", + $bindings + ); + + foreach ($rowsResult as $r) { + $results[] = $r; + } + } + } + + return $results; } private function binaryLookupKey(string $hash, int $collectionId): string @@ -441,6 +470,11 @@ final class BinaryHandler */ private function flushUpdatesMysql(array $updates, int $chunkSize): bool { + // Clamp the caller-provided chunk size to our hard upper bound so an + // unusually large config value cannot produce a 100 KB+ UNION ALL + // query that exhausts memory on either side of the connection. + $chunkSize = max(1, min($chunkSize, self::MAX_SQL_ROWS_PER_STATEMENT)); + foreach (array_chunk($updates, $chunkSize) as $chunk) { $selects = []; $bindings = []; diff --git a/app/Services/Binaries/CollectionHandler.php b/app/Services/Binaries/CollectionHandler.php index 835b7a0eb..d54b620e6 100644 --- a/app/Services/Binaries/CollectionHandler.php +++ b/app/Services/Binaries/CollectionHandler.php @@ -16,6 +16,13 @@ use Illuminate\Support\Facades\Log; */ final class CollectionHandler { + /** + * Hard upper bound on rows packed into a single SQL statement + * (multi-row INSERT, IN(...) lookup, etc.). Keeps the generated SQL + * and PDO parameter count bounded regardless of caller batch size. + */ + private const MAX_SQL_ROWS_PER_STATEMENT = 500; + private CollectionsCleaningService $collectionsCleaning; private XrefService $xrefService; @@ -237,10 +244,13 @@ final class CollectionHandler return; } - $rows = Collection::query() - ->whereIn('collectionhash', array_values($missing)) - ->pluck('xref', 'collectionhash') - ->all(); + $rows = []; + foreach (array_chunk(array_values($missing), self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $rows += Collection::query() + ->whereIn('collectionhash', $chunk) + ->pluck('xref', 'collectionhash') + ->all(); + } foreach ($missing as $collectionKey => $hash) { $this->existingXrefs[$collectionKey] = isset($rows[$hash]) ? (string) $rows[$hash] : null; @@ -282,11 +292,16 @@ final class CollectionHandler return []; } - return Collection::query() - ->whereIn('collectionhash', $hashes) - ->pluck('collectionhash') - ->mapWithKeys(static fn (string $hash): array => [$hash => true]) - ->all(); + $existing = []; + foreach (array_chunk($hashes, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $existing += Collection::query() + ->whereIn('collectionhash', $chunk) + ->pluck('collectionhash') + ->mapWithKeys(static fn (string $hash): array => [$hash => true]) + ->all(); + } + + return $existing; } /** @@ -310,7 +325,9 @@ final class CollectionHandler ]; } - DB::table('collections')->insertOrIgnore($rows); + foreach (array_chunk($rows, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + DB::table('collections')->insertOrIgnore($chunk); + } } /** @@ -319,31 +336,33 @@ final class CollectionHandler */ private function bulkInsertCollectionsMysql(array $rowsByCollectionKey, array $existingHashes): void { - $placeholders = []; - $bindings = []; - foreach ($rowsByCollectionKey as $row) { - $placeholders[] = '(?, ?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, NOW(), ?)'; - array_push( - $bindings, - $row['subject'], - $row['fromname'], - $row['unixtime'], - $row['xref'], - $row['groups_id'], - $row['totalfiles'], - $row['collectionhash'], - $row['collection_regexes_id'], - $row['noise'] + foreach (array_chunk(array_values($rowsByCollectionKey), self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = []; + $bindings = []; + foreach ($chunk as $row) { + $placeholders[] = '(?, ?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, NOW(), ?)'; + array_push( + $bindings, + $row['subject'], + $row['fromname'], + $row['unixtime'], + $row['xref'], + $row['groups_id'], + $row['totalfiles'], + $row['collectionhash'], + $row['collection_regexes_id'], + $row['noise'] + ); + } + + DB::statement( + 'INSERT INTO collections (subject, fromname, date, xref, groups_id, totalfiles, collectionhash, collection_regexes_id, dateadded, noise) VALUES ' + .implode(',', $placeholders) + .' ON DUPLICATE KEY UPDATE dateadded = NOW()', + $bindings ); } - DB::statement( - 'INSERT INTO collections (subject, fromname, date, xref, groups_id, totalfiles, collectionhash, collection_regexes_id, dateadded, noise) VALUES ' - .implode(',', $placeholders) - .' ON DUPLICATE KEY UPDATE dateadded = NOW()', - $bindings - ); - foreach ($rowsByCollectionKey as $row) { if ($row['xref_append'] !== '' && isset($existingHashes[$row['collectionhash']])) { DB::update('UPDATE collections SET xref = CONCAT(xref, ?, ?) WHERE collectionhash = ?', [ @@ -365,11 +384,16 @@ final class CollectionHandler return []; } - return Collection::query() - ->whereIn('collectionhash', $hashes) - ->pluck('id', 'collectionhash') - ->mapWithKeys(static fn (int|string $id, string $hash): array => [$hash => (int) $id]) - ->all(); + $resolved = []; + foreach (array_chunk($hashes, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $resolved += Collection::query() + ->whereIn('collectionhash', $chunk) + ->pluck('id', 'collectionhash') + ->mapWithKeys(static fn (int|string $id, string $hash): array => [$hash => (int) $id]) + ->all(); + } + + return $resolved; } /** diff --git a/app/Services/Binaries/HeaderStorageService.php b/app/Services/Binaries/HeaderStorageService.php index 9629f5acd..a68857870 100644 --- a/app/Services/Binaries/HeaderStorageService.php +++ b/app/Services/Binaries/HeaderStorageService.php @@ -54,9 +54,21 @@ final class HeaderStorageService $this->failedInserts = []; - $chunkSize = max(1, $this->config->partsChunkSize); - foreach (array_chunk($headers, $chunkSize) as $chunk) { + // Use the dedicated header chunk size, NOT partsChunkSize. The latter + // controls single-row part flushes and is normally much larger; using + // it here forces every collection/binary bulk INSERT and OR-clause + // SELECT to scale to thousands of rows per chunk, which exhausts PHP + // and MySQL memory. + $chunkSize = max(1, $this->config->headerChunkSize); + + // Walk the array with offset slicing instead of array_chunk() so we + // don't materialize every chunk simultaneously in memory. + $total = \count($headers); + $headers = array_values($headers); + for ($offset = 0; $offset < $total; $offset += $chunkSize) { + $chunk = \array_slice($headers, $offset, $chunkSize); $this->storeChunk($chunk, $groupMySQL, $addToPartRepair); + unset($chunk); } return array_values(array_unique($this->failedInserts)); diff --git a/app/Services/Binaries/HeaderStorageTransaction.php b/app/Services/Binaries/HeaderStorageTransaction.php index cd2cfc54a..30a9c0b4a 100644 --- a/app/Services/Binaries/HeaderStorageTransaction.php +++ b/app/Services/Binaries/HeaderStorageTransaction.php @@ -12,6 +12,8 @@ use Illuminate\Support\Facades\Log; */ final class HeaderStorageTransaction { + private const MAX_SQL_ROWS_PER_STATEMENT = 500; + private CollectionHandler $collectionHandler; private BinaryHandler $binaryHandler; @@ -126,18 +128,18 @@ final class HeaderStorageTransaction private function cleanupParts(): void { $ids = $this->binaryHandler->getInsertedIds(); - if (! empty($ids)) { - $placeholders = implode(',', array_fill(0, \count($ids), '?')); - DB::statement("DELETE FROM parts WHERE binaries_id IN ({$placeholders})", $ids); + foreach (array_chunk($ids, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), '?')); + DB::statement("DELETE FROM parts WHERE binaries_id IN ({$placeholders})", $chunk); } } private function cleanupBinaries(): void { $ids = $this->binaryHandler->getInsertedIds(); - if (! empty($ids)) { - $placeholders = implode(',', array_fill(0, \count($ids), '?')); - DB::statement("DELETE FROM binaries WHERE id IN ({$placeholders})", $ids); + foreach (array_chunk($ids, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), '?')); + DB::statement("DELETE FROM binaries WHERE id IN ({$placeholders})", $chunk); } } @@ -150,27 +152,30 @@ final class HeaderStorageTransaction $ids = ! empty($insertedIds) ? $insertedIds : $allIds; if (! empty($ids)) { - $placeholders = implode(',', array_fill(0, \count($ids), '?')); + foreach (array_chunk($ids, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), '?')); - // Remove parts and binaries referencing these collections, then collections - DB::statement( - "DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN ({$placeholders}))", - $ids - ); - DB::statement("DELETE FROM binaries WHERE collections_id IN ({$placeholders})", $ids); - DB::statement("DELETE FROM collections WHERE id IN ({$placeholders})", $ids); + DB::statement( + "DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN ({$placeholders}))", + $chunk + ); + DB::statement("DELETE FROM binaries WHERE collections_id IN ({$placeholders})", $chunk); + DB::statement("DELETE FROM collections WHERE id IN ({$placeholders})", $chunk); + } } elseif (! empty($hashes)) { - $placeholders = implode(',', array_fill(0, \count($hashes), '?')); + foreach (array_chunk($hashes, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), '?')); - DB::statement( - "DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ({$placeholders})))", - $hashes - ); - DB::statement( - "DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ({$placeholders}))", - $hashes - ); - DB::statement("DELETE FROM collections WHERE collectionhash IN ({$placeholders})", $hashes); + DB::statement( + "DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ({$placeholders})))", + $chunk + ); + DB::statement( + "DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ({$placeholders}))", + $chunk + ); + DB::statement("DELETE FROM collections WHERE collectionhash IN ({$placeholders})", $chunk); + } } else { // Fallback by noise marker DB::statement( diff --git a/app/Services/Binaries/PartHandler.php b/app/Services/Binaries/PartHandler.php index 5591734c8..a2f8d88d8 100644 --- a/app/Services/Binaries/PartHandler.php +++ b/app/Services/Binaries/PartHandler.php @@ -12,6 +12,14 @@ use Illuminate\Support\Facades\Log; */ final class PartHandler { + /** + * Hard upper bound on rows packed into a single SQL statement + * (multi-row INSERT, OR-clause SELECT). The public chunkSize controls + * when we flush; this constant guarantees the actual SQL we emit is + * never large enough to blow up PHP/MySQL memory. + */ + private const MAX_SQL_ROWS_PER_STATEMENT = 500; + /** @var array Pending parts to insert */ private array $parts = []; @@ -123,25 +131,31 @@ final class PartHandler */ private function insertChunk(array $parts): ?int { - $placeholders = []; - $bindings = []; $driver = DB::getDriverName(); - - foreach ($parts as $row) { - $placeholders[] = '(?,?,?,?,?)'; - $bindings[] = $row['binaries_id']; - $bindings[] = $row['number']; - $bindings[] = $row['messageid']; - $bindings[] = $row['partnumber']; - $bindings[] = $row['size']; - } - - $sql = $driver === 'sqlite' - ? 'INSERT OR IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders) - : 'INSERT IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders); + $totalInserted = 0; try { - return DB::affectingStatement($sql, $bindings); + foreach (array_chunk($parts, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = []; + $bindings = []; + + foreach ($chunk as $row) { + $placeholders[] = '(?,?,?,?,?)'; + $bindings[] = $row['binaries_id']; + $bindings[] = $row['number']; + $bindings[] = $row['messageid']; + $bindings[] = $row['partnumber']; + $bindings[] = $row['size']; + } + + $sql = $driver === 'sqlite' + ? 'INSERT OR IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders) + : 'INSERT IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders); + + $totalInserted += (int) DB::affectingStatement($sql, $bindings); + } + + return $totalInserted; } catch (\Throwable $e) { if (config('app.debug') === true) { Log::error('Parts chunk insert failed: '.$e->getMessage()); @@ -161,22 +175,30 @@ final class PartHandler return []; } - $clauses = []; - $bindings = []; + $keys = []; + // Group by binaries_id and use IN(...) per binary so the SELECT does + // not turn into thousands of OR clauses with two bindings each. + $numbersByBinary = []; foreach ($parts as $part) { - $clauses[] = '(binaries_id = ? AND number = ?)'; - $bindings[] = $part['binaries_id']; - $bindings[] = $part['number']; + $numbersByBinary[(int) $part['binaries_id']][] = $part['number']; } - $rows = DB::select( - 'SELECT binaries_id, number FROM parts WHERE '.implode(' OR ', $clauses), - $bindings - ); + foreach ($numbersByBinary as $binaryId => $numbers) { + $numbers = array_values(array_unique($numbers)); + foreach (array_chunk($numbers, self::MAX_SQL_ROWS_PER_STATEMENT) as $chunk) { + $placeholders = implode(',', array_fill(0, \count($chunk), '?')); + $bindings = $chunk; + $bindings[] = $binaryId; - $keys = []; - foreach ($rows as $row) { - $keys[$this->partKey((int) $row->binaries_id, (int) $row->number)] = true; + $rows = DB::select( + "SELECT binaries_id, number FROM parts WHERE number IN ({$placeholders}) AND binaries_id = ?", + $bindings + ); + + foreach ($rows as $row) { + $keys[$this->partKey((int) $row->binaries_id, (int) $row->number)] = true; + } + } } return $keys; diff --git a/tests/Feature/BinariesStorageInternalsTest.php b/tests/Feature/BinariesStorageInternalsTest.php index da122816c..562dc120a 100644 --- a/tests/Feature/BinariesStorageInternalsTest.php +++ b/tests/Feature/BinariesStorageInternalsTest.php @@ -128,7 +128,7 @@ class BinariesStorageInternalsTest extends TestCase { $this->createHeaderStorageTables('CHECK(size < 500)'); - $service = new HeaderStorageService($this->deterministicCollectionHandler(), config: new BinariesConfig(partsChunkSize: 2)); + $service = new HeaderStorageService($this->deterministicCollectionHandler(), config: new BinariesConfig(partsChunkSize: 2, headerChunkSize: 2)); $failed = $service->store([ $this->parsedHeader(301, 1, 'Chunk.One', 100), $this->parsedHeader(302, 2, 'Chunk.One', 100),