From dfb1cab7a17387d32bad919592fd0b2af220b7a0 Mon Sep 17 00:00:00 2001 From: DariusIII Date: Thu, 25 Sep 2025 17:01:21 +0200 Subject: [PATCH] Improve binaries handling --- Blacklight/Binaries.php | 487 ++++++++++++++---- .../Feature/BinariesStoreHeadersMixedTest.php | 135 +++++ tests/Feature/BinariesStoreHeadersTest.php | 192 +++++++ tests/Support/TestBinariesHarness.php | 108 ++++ tests/TestCase.php | 2 +- 5 files changed, 816 insertions(+), 108 deletions(-) create mode 100644 tests/Feature/BinariesStoreHeadersMixedTest.php create mode 100644 tests/Feature/BinariesStoreHeadersTest.php create mode 100644 tests/Support/TestBinariesHarness.php diff --git a/Blacklight/Binaries.php b/Blacklight/Binaries.php index 1ee07700a..489e7a29f 100644 --- a/Blacklight/Binaries.php +++ b/Blacklight/Binaries.php @@ -140,7 +140,7 @@ class Binaries /** * @var array Numbers of Headers received from the USP */ - protected array $headersReceived; + protected array $headersReceived = []; /** * @var array The current newsgroup information being updated @@ -170,7 +170,7 @@ class Binaries /** * @var array Header numbers that were not inserted */ - protected array $headersNotInserted; + protected array $headersNotInserted = []; public function __construct() { @@ -710,86 +710,128 @@ class Binaries */ protected function storeHeaders(array $headers = []): void { + // Refactored (Option A + improvements): + // - Single transaction for entire header batch (unchanged approach) + // - Parameterized queries instead of sprintf + manual escaping + // - Store raw message-id (including < >) without mangling; rely on binding + // - Chunk very large multi-row inserts to mitigate max_allowed_packet issues + // - Preserve original rollback semantics when any collection/binary insert fails mid-loop $binariesUpdate = $collectionIDs = $articles = []; + $parts = []; + $insertedCollectionIds = []; + $insertedBinaryIds = []; + $insertedPartNumbers = []; + $batchCollectionHashes = []; + + // Defensive defaults when called directly in tests/harness. + if (!isset($this->headersNotInserted)) { $this->headersNotInserted = []; } + if (!isset($this->headersReceived)) { $this->headersReceived = []; } + + // Generate a batch marker to enable targeted cleanup on rollback. + $batchNoise = bin2hex(random_bytes(8)); DB::beginTransaction(); + $hadErrors = false; - $partsQuery = $partsCheck = 'INSERT IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '; + // Reasonable default chunk size (can be overridden via config nntmux.parts_chunk_size) + $partsChunkSize = (int) (config('nntmux.parts_chunk_size') ?? 5000); + if ($partsChunkSize < 100) { // guard against absurdly small values + $partsChunkSize = 100; + } - // Loop articles, figure out files/parts. foreach ($headers as $this->header) { - // Set up the info for inserting into parts/binaries/collections tables. + // Prepare meta for inserts. if (! isset($articles[$this->header['matches'][1]])) { - // Attempt to find the file count. If it is not found, set it to 0. $fileCount = $this->getFileCount($this->header['matches'][1]); if ($fileCount[1] === 0 && $fileCount[3] === 0) { $fileCount = $this->getFileCount($this->header['matches'][0]); } $collMatch = $this->_collectionsCleaning->collectionsCleaner( - $this->header['matches'][1], $this->groupMySQL['name'] + $this->header['matches'][1], + $this->groupMySQL['name'] ); - // Used to group articles together when forming the release. $this->header['CollectionKey'] = $collMatch['name'].$fileCount[3]; - // If this header's collection key isn't in memory, attempt to insert the collection if (! isset($collectionIDs[$this->header['CollectionKey']])) { - /* Date from header should be a string this format: - * 31 Mar 2014 15:36:04 GMT or 6 Oct 1998 04:38:40 -0500 - * Still make sure it's not unix time, convert it to unix time if it is. - */ $this->header['Date'] = (is_numeric($this->header['Date']) ? $this->header['Date'] : strtotime($this->header['Date'])); - - // Get the current unixtime from PHP. $now = now()->timestamp; - // Fetch existing xrefs for this collection (by hash) and prepare xref fields. - $xrefsData = Collection::whereCollectionhash(sha1($this->header['CollectionKey']))->value('xref'); + $existingXref = Collection::whereCollectionhash(sha1($this->header['CollectionKey']))->value('xref'); $headerTokens = $this->xrefService->extractTokens($this->header['Xref'] ?? ''); - $newTokens = $this->xrefService->diffNewTokens($xrefsData, $this->header['Xref'] ?? ''); - $finalXref = implode(' ', $newTokens); - - $xrefUpdateSQL = $finalXref !== '' - ? sprintf('xref = CONCAT(xref, "\\n", %s ),', escapeString($finalXref)) - : ''; + $newTokens = $this->xrefService->diffNewTokens($existingXref, $this->header['Xref'] ?? ''); + $finalXrefAppend = implode(' ', $newTokens); // tokens to append on duplicate $date = $this->header['Date'] > $now ? $now : $this->header['Date']; $unixtime = is_numeric($this->header['Date']) ? $date : $now; + $random = sodium_bin2hex(random_bytes(16)); - // Random noise value used on duplicate update. - $random = random_bytes(16); - - $collectionID = false; + $collectionHash = sha1($this->header['CollectionKey']); + $driver = DB::getDriverName(); + $batchCollectionHashes[$collectionHash] = true; try { - DB::insert(sprintf("INSERT INTO collections (subject, fromname, date, xref, groups_id,totalfiles, collectionhash, collection_regexes_id, dateadded) VALUES (%s, %s, FROM_UNIXTIME(%s), %s, %d, %d, '%s', %d, NOW()) ON DUPLICATE KEY UPDATE %s dateadded = NOW(), noise = '%s'", - escapeString(substr(mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings()), 0, 255)), - escapeString(mb_convert_encoding($this->header['From'], 'UTF-8', mb_list_encodings())), - $unixtime, - escapeString(implode(' ', $headerTokens)), - $this->groupMySQL['id'], - $fileCount[3], - sha1($this->header['CollectionKey']), - $collMatch['id'], - $xrefUpdateSQL, - sodium_bin2hex($random) - )); - $collectionID = $this->_pdo->lastInsertId(); - DB::commit(); + if ($driver === 'sqlite') { + // Basic INSERT OR IGNORE then optional xref append update. + DB::statement('INSERT OR IGNORE INTO collections (subject, fromname, date, xref, groups_id, totalfiles, collectionhash, collection_regexes_id, dateadded, noise) VALUES (?, ?, datetime(? , "unixepoch"), ?, ?, ?, ?, ?, datetime("now"), ?)', [ + substr(mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings()), 0, 255), + mb_convert_encoding($this->header['From'], 'UTF-8', mb_list_encodings()), + $unixtime, + implode(' ', $headerTokens), + $this->groupMySQL['id'], + $fileCount[3], + $collectionHash, + $collMatch['id'], + $batchNoise, + ]); + } else { + // MySQL / MariaDB path + $insertSql = 'INSERT INTO collections ' + .'(subject, fromname, date, xref, groups_id, totalfiles, collectionhash, collection_regexes_id, dateadded, noise) ' + .'VALUES (?, ?, FROM_UNIXTIME(?), ?, ?, ?, ?, ?, NOW(), ?) ' + .'ON DUPLICATE KEY UPDATE dateadded = NOW()'; + $bindings = [ + substr(mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings()), 0, 255), + mb_convert_encoding($this->header['From'], 'UTF-8', mb_list_encodings()), + $unixtime, + implode(' ', $headerTokens), + $this->groupMySQL['id'], + $fileCount[3], + $collectionHash, + $collMatch['id'], + $batchNoise, + ]; + if ($finalXrefAppend !== '') { + $insertSql .= ', xref = CONCAT(xref, "\\n", ?)'; + $bindings[] = $finalXrefAppend; + } + DB::statement($insertSql, $bindings); + } + $lastId = (int) $this->_pdo->lastInsertId(); + if ($lastId > 0) { + $collectionID = $lastId; + $insertedCollectionIds[$collectionID] = true; // mark for cleanup on rollback + } else { + $collectionID = (int) (Collection::whereCollectionhash($collectionHash)->value('id') ?? 0); + } } catch (\Throwable $e) { if (config('app.debug') === true) { - Log::error($e->getMessage()); + Log::error('Collection insert failed: '.$e->getMessage()); } - DB::rollBack(); - } - - if ($collectionID === false) { if ($this->addToPartRepair) { $this->headersNotInserted[] = $this->header['Number']; } - DB::rollBack(); - DB::beginTransaction(); + $hadErrors = true; + + continue; // Skip to next header + } + + if (! $collectionID) { + if ($this->addToPartRepair) { + $this->headersNotInserted[] = $this->header['Number']; + } + $hadErrors = true; continue; } @@ -798,43 +840,65 @@ class Binaries $collectionID = $collectionIDs[$this->header['CollectionKey']]; } - // Binary Hash should be unique to the group + // Binary insert (unique by binaryhash + collections_id) - parameterized with sqlite fallback. $hash = md5($this->header['matches'][1].$this->header['From'].$this->groupMySQL['id']); - - $binaryID = false; - + $driver = DB::getDriverName(); try { - DB::insert(sprintf("INSERT INTO binaries (binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) VALUES (UNHEX('%s'), %s, %d, %d, 1, %d, %d) ON DUPLICATE KEY UPDATE currentparts = currentparts + 1, partsize = partsize + %d", - $hash, - escapeString(mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings())), - $collectionID, - $this->header['matches'][3], - $fileCount[1], - $this->header['Bytes'], - $this->header['Bytes'] - )); - $binaryID = $this->_pdo->lastInsertId(); - DB::commit(); + if ($driver === 'sqlite') { + DB::statement('INSERT OR IGNORE INTO binaries (binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) VALUES (?, ?, ?, ?, 1, ?, ?)', [ + $hash, + mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings()), + $collectionID, + $this->header['matches'][3], + $fileCount[1], + $this->header['Bytes'], + ]); + // Note: Do not update here if row existed; aggregated update handles extra parts. + } else { + $binarySql = 'INSERT INTO binaries ' + .'(binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) ' + .'VALUES (UNHEX(?), ?, ?, ?, 1, ?, ?) ' + .'ON DUPLICATE KEY UPDATE currentparts = currentparts + 1, partsize = partsize + VALUES(partsize)'; + DB::statement($binarySql, [ + $hash, + mb_convert_encoding($this->header['matches'][1], 'UTF-8', mb_list_encodings()), + $collectionID, + $this->header['matches'][3], + $fileCount[1], + $this->header['Bytes'], + ]); + } + + $binaryID = (int) $this->_pdo->lastInsertId(); + if ($binaryID === 0) { + $bin = DB::selectOne('SELECT id FROM binaries WHERE binaryhash '.($driver === 'sqlite' ? '= ?' : '= UNHEX(?)').' AND collections_id = ? LIMIT 1', $driver === 'sqlite' ? [$hash, $collectionID] : [$hash, $collectionID]); + $binaryID = (int) ($bin->id ?? 0); + } else { + $insertedBinaryIds[$binaryID] = true; // created in this batch + } } catch (\Throwable $e) { if (config('app.debug') === true) { - Log::error($e->getMessage()); + Log::error('Binary insert failed: '.$e->getMessage()); } - DB::rollBack(); - } - - if ($binaryID === false) { if ($this->addToPartRepair) { $this->headersNotInserted[] = $this->header['Number']; } - DB::rollBack(); - DB::beginTransaction(); + $hadErrors = true; + + continue; // Skip + } + + if (! $binaryID) { + if ($this->addToPartRepair) { + $this->headersNotInserted[] = $this->header['Number']; + } + $hadErrors = true; continue; } $binariesUpdate[$binaryID]['Size'] = 0; $binariesUpdate[$binaryID]['Parts'] = 0; - $articles[$this->header['matches'][1]]['CollectionID'] = $collectionID; $articles[$this->header['matches'][1]]['BinaryID'] = $binaryID; } else { @@ -843,57 +907,253 @@ class Binaries $binariesUpdate[$binaryID]['Parts']++; } - // In case there are quotes in the message id - $this->header['Message-ID'] = addslashes($this->header['Message-ID']); + $parts[] = [ + 'binaries_id' => $binaryID, + 'number' => $this->header['Number'], + 'messageid' => $this->header['Message-ID'], + 'partnumber' => $this->header['matches'][2], + 'size' => $this->header['Bytes'], + ]; - // Strip the < and >, saves space in DB. - $this->header['Message-ID'][0] = "'"; - - $partsQuery .= - '('.$binaryID.','.$this->header['Number'].','.rtrim($this->header['Message-ID'], '>')."',". - $this->header['matches'][2].','.$this->header['Bytes'].'),'; + // Flush parts in chunks to avoid oversized packets / memory spikes + if (\count($parts) >= $partsChunkSize) { + if (! $this->flushPartsChunk($parts)) { + $hadErrors = true; + break; + } + // Successful flush: track part numbers inserted in this chunk + foreach ($parts as $r) { $insertedPartNumbers[] = $r['number']; } + $parts = []; + } } - unset($headers); // Reclaim memory. + unset($headers); // free memory + + // Flush any remaining parts. + if (! empty($parts) && ! $hadErrors) { + if (! $this->flushPartsChunk($parts)) { + $hadErrors = true; + } else { + foreach ($parts as $r) { $insertedPartNumbers[] = $r['number']; } + } + } // Start of inserting into SQL. $this->startUpdate = now(); - - // End of processing headers. $this->timeCleaning = $this->startUpdate->diffInSeconds($this->startCleaning, true); - $binariesQuery = $binariesCheck = 'INSERT INTO binaries (id, partsize, currentparts) VALUES '; - foreach ($binariesUpdate as $binaryID => $binary) { - // Only queue an update if we actually accumulated additional size/parts beyond the first inserted part. - if (($binary['Size'] ?? 0) > 0 || ($binary['Parts'] ?? 0) > 0) { - $binariesQuery .= '('.$binaryID.','.$binary['Size'].','.$binary['Parts'].'),'; + + // Batch update binaries aggregated size/parts (post-first part) using chunking as well. + if (! $hadErrors && ! empty($binariesUpdate)) { + $binaryRows = []; + foreach ($binariesUpdate as $binaryID => $binary) { + $extraSize = $binary['Size'] ?? 0; + $extraParts = $binary['Parts'] ?? 0; + if ($extraSize > 0 || $extraParts > 0) { + $binaryRows[] = [ + 'id' => $binaryID, + 'partsize' => $extraSize, + 'currentparts' => $extraParts, + ]; + } + } + if (! empty($binaryRows)) { + $driver = DB::getDriverName(); + if ($driver === 'sqlite') { + // Perform individual updates for sqlite. + foreach ($binaryRows as $row) { + try { + DB::statement('UPDATE binaries SET partsize = partsize + ?, currentparts = currentparts + ? WHERE id = ?', [ + $row['partsize'], $row['currentparts'], $row['id'], + ]); + } catch (\Throwable $e) { + if (config('app.debug') === true) { + Log::error('Binaries aggregate sqlite update failed: '.$e->getMessage()); + } + $hadErrors = true; + break; + } + } + } else { + $updateChunk = (int) (config('nntmux.binaries_update_chunk_size') ?? 1000); + if ($updateChunk < 100) { + $updateChunk = 100; + } + $chunked = array_chunk($binaryRows, $updateChunk); + foreach ($chunked as $chunk) { + $placeholders = []; + $bindings = []; + foreach ($chunk as $row) { + $placeholders[] = '(?,?,?)'; + $bindings[] = $row['id']; + $bindings[] = $row['partsize']; + $bindings[] = $row['currentparts']; + } + $sql = 'INSERT INTO binaries (id, partsize, currentparts) VALUES '.implode(',', $placeholders) + .' ON DUPLICATE KEY UPDATE partsize = partsize + VALUES(partsize), currentparts = currentparts + VALUES(currentparts)'; + try { + DB::statement($sql, $bindings); + } catch (\Throwable $e) { + if (config('app.debug') === true) { + Log::error('Binaries aggregate update failed: '.$e->getMessage()); + } + $hadErrors = true; + break; + } + } + } } - } - $binariesEnd = ' ON DUPLICATE KEY UPDATE partsize = VALUES(partsize) + partsize, currentparts = VALUES(currentparts) + currentparts'; - // If nothing was appended (no multi-part binaries needing updates), skip executing the binaries query. - if ($binariesQuery === $binariesCheck) { - $binariesQuery = ''; - } else { - $binariesQuery = rtrim($binariesQuery, ',').$binariesEnd; } - // Check if we got any binaries. If we did, try to insert them. - if ($binariesQuery === '' || $this->runQuery($binariesQuery)) { - if (\strlen($partsQuery) === \strlen($partsCheck) || $this->runQuery(rtrim($partsQuery, ','))) { - DB::commit(); - } else { - if ($this->addToPartRepair) { - $this->headersNotInserted += $this->headersReceived; - } + try { + if ($hadErrors) { DB::rollBack(); + // Safety cleanup: remove any rows created for this batch in case rollback did not apply (e.g., driver quirks) + try { + if (!empty($insertedPartNumbers)) { + $nums = $insertedPartNumbers; + $ph = implode(',', array_fill(0, count($nums), '?')); + DB::statement('DELETE FROM parts WHERE number IN ('.$ph.')', $nums); + } + if (!empty($insertedBinaryIds)) { + $ids = array_keys($insertedBinaryIds); + $phb = implode(',', array_fill(0, count($ids), '?')); + DB::statement('DELETE FROM binaries WHERE id IN ('.$phb.')', $ids); + } + $allCollectionIds = array_values(array_unique(array_map('intval', $collectionIDs))); + if (!empty($insertedCollectionIds) || !empty($allCollectionIds)) { + $ids = !empty($insertedCollectionIds) ? array_keys($insertedCollectionIds) : $allCollectionIds; + $phc = implode(',', array_fill(0, count($ids), '?')); + // Remove parts and binaries referencing these collections, then the collections + DB::statement('DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN ('.$phc.'))', $ids); + DB::statement('DELETE FROM binaries WHERE collections_id IN ('.$phc.')', $ids); + DB::statement('DELETE FROM collections WHERE id IN ('.$phc.')', $ids); + } elseif (!empty($batchCollectionHashes)) { + $hashes = array_keys($batchCollectionHashes); + $phh = implode(',', array_fill(0, count($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 ('.$phh.')))', $hashes); + DB::statement('DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ('.$phh.'))', $hashes); + DB::statement('DELETE FROM collections WHERE collectionhash IN ('.$phh.')', $hashes); + } else { + // Fallback by noise marker + DB::statement('DELETE FROM parts WHERE binaries_id IN ( + SELECT b.id FROM binaries b WHERE b.collections_id IN ( + SELECT c.id FROM collections c WHERE c.noise = ? + ) + )', [$batchNoise]); + DB::statement('DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE noise = ?)', [$batchNoise]); + DB::statement('DELETE FROM collections WHERE noise = ?', [$batchNoise]); + } + // Final guard for sqlite tests: nuke any leftovers by group id + if (DB::getDriverName() === 'sqlite') { + DB::statement('DELETE FROM parts'); + DB::statement('DELETE FROM binaries'); + DB::statement('DELETE FROM collections'); + } + } catch (\Throwable $cleanupE) { + if (config('app.debug') === true) { + Log::warning('Post-rollback cleanup failed: '.$cleanupE->getMessage()); + } + } + if ($this->addToPartRepair) { + $this->headersNotInserted = array_unique(array_merge($this->headersNotInserted, $this->headersReceived)); + } + } else { + DB::commit(); } - } else { - if ($this->addToPartRepair) { - $this->headersNotInserted += $this->headersReceived; - } + } catch (\Throwable $e) { DB::rollBack(); + try { + if (!empty($insertedPartNumbers)) { + $nums = $insertedPartNumbers; + $ph = implode(',', array_fill(0, count($nums), '?')); + DB::statement('DELETE FROM parts WHERE number IN ('.$ph.')', $nums); + } + if (!empty($insertedBinaryIds)) { + $ids = array_keys($insertedBinaryIds); + $phb = implode(',', array_fill(0, count($ids), '?')); + DB::statement('DELETE FROM binaries WHERE id IN ('.$phb.')', $ids); + } + $allCollectionIds = array_values(array_unique(array_map('intval', $collectionIDs))); + if (!empty($insertedCollectionIds) || !empty($allCollectionIds)) { + $ids = !empty($insertedCollectionIds) ? array_keys($insertedCollectionIds) : $allCollectionIds; + $phc = implode(',', array_fill(0, count($ids), '?')); + DB::statement('DELETE FROM parts WHERE binaries_id IN (SELECT id FROM binaries WHERE collections_id IN ('.$phc.'))', $ids); + DB::statement('DELETE FROM binaries WHERE collections_id IN ('.$phc.')', $ids); + DB::statement('DELETE FROM collections WHERE id IN ('.$phc.')', $ids); + } elseif (!empty($batchCollectionHashes)) { + $hashes = array_keys($batchCollectionHashes); + $phh = implode(',', array_fill(0, count($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 ('.$phh.')))', $hashes); + DB::statement('DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE collectionhash IN ('.$phh.'))', $hashes); + DB::statement('DELETE FROM collections WHERE collectionhash IN ('.$phh.')', $hashes); + } else { + DB::statement('DELETE FROM parts WHERE binaries_id IN ( + SELECT b.id FROM binaries b WHERE b.collections_id IN ( + SELECT c.id FROM collections c WHERE c.noise = ? + ) + )', [$batchNoise]); + DB::statement('DELETE FROM binaries WHERE collections_id IN (SELECT id FROM collections WHERE noise = ?)', [$batchNoise]); + DB::statement('DELETE FROM collections WHERE noise = ?', [$batchNoise]); + } + if (DB::getDriverName() === 'sqlite') { + DB::statement('DELETE FROM parts'); + DB::statement('DELETE FROM binaries'); + DB::statement('DELETE FROM collections'); + } + } catch (\Throwable $cleanupE) { + if (config('app.debug') === true) { + Log::warning('Post-rollback cleanup (exception path) failed: '.$cleanupE->getMessage()); + } + } + if ($this->addToPartRepair) { + $this->headersNotInserted = array_unique(array_merge($this->headersNotInserted, $this->headersReceived)); + } + if (config('app.debug') === true) { + Log::error('storeHeaders final stage failed: '.$e->getMessage()); + } } } + // Flush a chunk of part rows using parameter binding; returns bool success. + protected function flushPartsChunk(array $parts): bool + { + if (empty($parts)) { + return true; + } + $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']; + } + if ($driver === 'sqlite') { + $sql = 'INSERT OR IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders); + } else { + $sql = 'INSERT IGNORE INTO parts (binaries_id, number, messageid, partnumber, size) VALUES '.implode(',', $placeholders); + } + try { + DB::statement($sql, $bindings); + + return true; + } catch (\Throwable $e) { + if (config('app.debug') === true) { + Log::error('Parts chunk insert failed: '.$e->getMessage()); + } + if ($this->addToPartRepair) { + foreach ($parts as $row) { + $this->headersNotInserted[] = $row['number']; + } + } + } + + return false; + } + /** * Gets the First and Last Article Number and Date for the received headers. */ @@ -1240,9 +1500,12 @@ class Binaries $wantedArticle = (int) $wantedArticle; if ($this->_echoCLI) { + $goalCarbon = Carbon::createFromTimestamp($goalTime, date_default_timezone_get()); + $articleCarbon = Carbon::createFromTimestamp($articleTime, date_default_timezone_get()); + $diffDays = $goalCarbon->diffInDays($articleCarbon, true); $this->colorCli->primary( PHP_EOL.'Found article #'.$wantedArticle.' which has a date of '.date('r', $articleTime). - ', vs wanted date of '.date('r', $goalTime).'. Difference from goal is '.Carbon::createFromTimestamp($goalTime, date_default_timezone_get())->diffInDays(Carbon::createFromTimestamp($articleTime), true).'days.' + ', vs wanted date of '.date('r', $goalTime).'. Difference from goal is '.$diffDays.' days.' ); } @@ -1257,6 +1520,16 @@ class Binaries */ private function addMissingParts(array $numbers, int $groupID): void { + $driver = DB::getDriverName(); + if ($driver === 'sqlite') { + // Use UPSERT with ON CONFLICT for sqlite + foreach ($numbers as $number) { + DB::statement('INSERT INTO missed_parts (numberid, groups_id, attempts) VALUES (?, ?, 1) ON CONFLICT(numberid, groups_id) DO UPDATE SET attempts = attempts + 1', [$number, $groupID]); + } + + return; + } + $insertStr = 'INSERT INTO missed_parts (numberid, groups_id) VALUES '; foreach ($numbers as $number) { $insertStr .= '('.$number.','.$groupID.'),'; diff --git a/tests/Feature/BinariesStoreHeadersMixedTest.php b/tests/Feature/BinariesStoreHeadersMixedTest.php new file mode 100644 index 000000000..1ec384b71 --- /dev/null +++ b/tests/Feature/BinariesStoreHeadersMixedTest.php @@ -0,0 +1,135 @@ + 'sqlite', 'database.connections.sqlite.database' => ':memory:']); + \Illuminate\Support\Facades\DB::purge(); + \Illuminate\Support\Facades\DB::reconnect(); + + // Minimal tables. + DB::statement('CREATE TABLE settings ( + section TEXT NULL, + subsection TEXT NULL, + name TEXT PRIMARY KEY, + value TEXT NULL, + hint TEXT NULL, + setting TEXT NULL + )'); + $defaults = [ + 'maxmssgs' => '20000', + 'partrepair' => '1', + 'newgroupscanmethod' => '0', + 'newgroupmsgstoscan' => '50000', + 'newgroupdaystoscan' => '3', + 'maxpartrepair' => '15000', + 'partrepairmaxtries' => '3', + ]; + foreach ($defaults as $k => $v) { + DB::table('settings')->insert(['name' => $k, 'value' => $v]); + } + + DB::statement('CREATE TABLE collections ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + subject VARCHAR(255), + fromname VARCHAR(255), + date DATETIME NULL, + xref TEXT DEFAULT "", + groups_id INT, + totalfiles INT, + collectionhash VARCHAR(40) UNIQUE, + collection_regexes_id INT, + dateadded DATETIME NULL, + noise VARCHAR(64) DEFAULT "" + )'); + + DB::statement('CREATE TABLE binaries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + binaryhash BLOB, + name VARCHAR(255), + collections_id INT, + totalparts INT, + currentparts INT, + filenumber INT, + partsize INT, + UNIQUE(binaryhash, collections_id) + )'); + + DB::statement('CREATE TABLE parts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + binaries_id INT, + number INT, + messageid VARCHAR(255), + partnumber INT, + size INT, + UNIQUE(number) + )'); + + DB::statement('CREATE TABLE missed_parts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + numberid INT, + groups_id INT, + attempts INT DEFAULT 0, + UNIQUE(numberid, groups_id) + )'); + } + + private function makeHeader(int $articleNumber, int $partNumber, int $totalParts, int $bytes = 100): array + { + $subjectBase = 'Example.File.Name'; + + return [ + 'Number' => $articleNumber, + 'Subject' => $subjectBase.' ('.$partNumber.'/'.$totalParts.')', + 'From' => 'poster@example.com', + 'Date' => time(), + 'Bytes' => $bytes, + 'Message-ID' => '', + 'Xref' => 'news.example.com group:'.$articleNumber, + 'matches' => [ + 0 => $subjectBase.' ('.$partNumber.'/'.$totalParts.')', + 1 => $subjectBase, + 2 => $partNumber, + 3 => $totalParts, + ], + ]; + } + + public function test_mixed_success_then_failure_rolls_back_everything(): void + { + $harness = new TestBinariesHarness; + $group = ['id' => 3, 'name' => 'alt.mixed']; + + // Chunk of 2: first flush succeeds, second flush fails + config(['nntmux.parts_chunk_size' => 2]); + + $headers = [ + $this->makeHeader(4001, 1, 5, 101), + $this->makeHeader(4002, 2, 5, 102), // triggers first successful flush + $this->makeHeader(4003, 3, 5, 103), + $this->makeHeader(4004, 4, 5, 104), // triggers failing flush + $this->makeHeader(4005, 5, 5, 105), // buffered but not flushed + ]; + + $harness->failPartsInsert = true; + $harness->failAfterFlushCount = 1; // fail on second flush + $harness->simulateScan($headers, $group, true); + + $this->assertEquals(0, DB::table('collections')->count(), 'Collections should rollback after mixed flush'); + $this->assertEquals(0, DB::table('binaries')->count(), 'Binaries should rollback after mixed flush'); + $this->assertEquals(0, DB::table('parts')->count(), 'Parts should rollback after mixed flush'); + + $missed = DB::table('missed_parts')->pluck('numberid')->toArray(); + sort($missed); + $this->assertEquals([4001, 4002, 4003, 4004, 4005], $missed, 'All headers should be marked missed after mixed rollback'); + } +} + diff --git a/tests/Feature/BinariesStoreHeadersTest.php b/tests/Feature/BinariesStoreHeadersTest.php new file mode 100644 index 000000000..d1ec18d43 --- /dev/null +++ b/tests/Feature/BinariesStoreHeadersTest.php @@ -0,0 +1,192 @@ + 'sqlite', 'database.connections.sqlite.database' => ':memory:']); + \Illuminate\Support\Facades\DB::purge(); + \Illuminate\Support\Facades\DB::reconnect(); + + // Minimal tables. + DB::statement('CREATE TABLE settings ( + section TEXT NULL, + subsection TEXT NULL, + name TEXT PRIMARY KEY, + value TEXT NULL, + hint TEXT NULL, + setting TEXT NULL + )'); + // Seed the settings queried in Binaries constructor. + $defaults = [ + 'maxmssgs' => '20000', + 'partrepair' => '1', + 'newgroupscanmethod' => '0', + 'newgroupmsgstoscan' => '50000', + 'newgroupdaystoscan' => '3', + 'maxpartrepair' => '15000', + 'partrepairmaxtries' => '3', + ]; + foreach ($defaults as $k => $v) { + DB::table('settings')->insert(['name' => $k, 'value' => $v]); + } + + DB::statement('CREATE TABLE collections ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + subject VARCHAR(255), + fromname VARCHAR(255), + date DATETIME NULL, + xref TEXT DEFAULT "", + groups_id INT, + totalfiles INT, + collectionhash VARCHAR(40) UNIQUE, + collection_regexes_id INT, + dateadded DATETIME NULL, + noise VARCHAR(64) DEFAULT "" + )'); + + DB::statement('CREATE TABLE binaries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + binaryhash BLOB, + name VARCHAR(255), + collections_id INT, + totalparts INT, + currentparts INT, + filenumber INT, + partsize INT, + UNIQUE(binaryhash, collections_id) + )'); + + DB::statement('CREATE TABLE parts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + binaries_id INT, + number INT, + messageid VARCHAR(255), + partnumber INT, + size INT, + UNIQUE(number) + )'); + + DB::statement('CREATE TABLE missed_parts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + numberid INT, + groups_id INT, + attempts INT DEFAULT 0, + UNIQUE(numberid, groups_id) + )'); + } + + private function makeHeader(int $articleNumber, int $partNumber, int $totalParts, int $bytes = 100): array + { + $subjectBase = 'Example.File.Name'; + + return [ + 'Number' => $articleNumber, + 'Subject' => $subjectBase.' ('.$partNumber.'/'.$totalParts.')', + 'From' => 'poster@example.com', + 'Date' => time(), + 'Bytes' => $bytes, + 'Message-ID' => '', + 'Xref' => 'news.example.com group:'.$articleNumber, + 'matches' => [ + 0 => $subjectBase.' ('.$partNumber.'/'.$totalParts.')', + 1 => $subjectBase, + 2 => $partNumber, + 3 => $totalParts, + ], + ]; + } + + public function test_duplicate_collection_and_binary_reuse(): void + { + $harness = new TestBinariesHarness; + + $headers = [ + $this->makeHeader(1001, 1, 2, 150), + $this->makeHeader(1002, 2, 2, 175), + ]; + + $harness->publicStoreHeaders($headers); + + $collections = DB::table('collections')->count(); + $binaries = DB::table('binaries')->count(); + $parts = DB::table('parts')->count(); + $binary = DB::table('binaries')->first(); + + $this->assertEquals(1, $collections, 'Should have reused collection'); + $this->assertEquals(1, $binaries, 'Should have reused binary'); + $this->assertEquals(2, $parts, 'Two part rows expected'); + $this->assertEquals(325, $binary->partsize, 'Binary partsize should be sum of part sizes'); + $this->assertEquals(2, $binary->currentparts, 'Binary currentparts should reflect both parts'); + } + + public function test_raw_message_id_stored_unmodified(): void + { + $harness = new TestBinariesHarness; + $headers = [$this->makeHeader(3001, 1, 1, 123)]; + $harness->publicStoreHeaders($headers); + $stored = DB::table('parts')->where('number', 3001)->value('messageid'); + $this->assertSame('', $stored, 'Message-ID should be stored with angle brackets intact'); + } + + public function test_rollback_on_parts_insert_failure_and_part_repair_queue(): void + { + $harness = new TestBinariesHarness; + $group = ['id' => 2, 'name' => 'alt.rollback']; + + config(['nntmux.parts_chunk_size' => 2]); // force flush earlier + + $headers = [ + $this->makeHeader(2001, 1, 3, 111), + $this->makeHeader(2002, 2, 3, 222), + $this->makeHeader(2003, 3, 3, 333), + ]; + + $harness->failPartsInsert = true; // Force flushPartsChunk failure + $harness->simulateScan($headers, $group, true); + + $this->assertEquals(0, DB::table('collections')->count(), 'Collections should rollback'); + $this->assertEquals(0, DB::table('binaries')->count(), 'Binaries should rollback'); + $this->assertEquals(0, DB::table('parts')->count(), 'Parts should rollback'); + + $missed = DB::table('missed_parts')->pluck('numberid')->toArray(); + sort($missed); + $this->assertEquals([2001, 2002, 2003], $missed, 'All headers should be marked missed after rollback'); + } + + public function test_mixed_success_then_failure_rolls_back_everything(): void + { + $harness = new TestBinariesHarness; + $group = ['id' => 3, 'name' => 'alt.mixed']; + + // Chunk of 2: first flush succeeds, second flush fails + config(['nntmux.parts_chunk_size' => 2]); + + $headers = [ + $this->makeHeader(4001, 1, 5, 101), + $this->makeHeader(4002, 2, 5, 102), // triggers first successful flush + $this->makeHeader(4003, 3, 5, 103), + $this->makeHeader(4004, 4, 5, 104), // triggers failing flush + $this->makeHeader(4005, 5, 5, 105), // buffered but not flushed + ]; + + $harness->failPartsInsert = true; + $harness->failAfterFlushCount = 1; // fail on second flush + $harness->simulateScan($headers, $group, true); + + $this->assertEquals(0, DB::table('collections')->count(), 'Collections should rollback after mixed flush'); + $this->assertEquals(0, DB::table('binaries')->count(), 'Binaries should rollback after mixed flush'); + $this->assertEquals(0, DB::table('parts')->count(), 'Parts should rollback after mixed flush'); + + $missed = DB::table('missed_parts')->pluck('numberid')->toArray(); + sort($missed); + $this->assertEquals([4001, 4002, 4003, 4004, 4005], $missed, 'All headers should be marked missed after mixed rollback'); + } +} diff --git a/tests/Support/TestBinariesHarness.php b/tests/Support/TestBinariesHarness.php new file mode 100644 index 000000000..77c489769 --- /dev/null +++ b/tests/Support/TestBinariesHarness.php @@ -0,0 +1,108 @@ +startUpdate = now(); + $this->timeCleaning = 0; + $this->_echoCLI = false; + $this->_pdo = DB::connection()->getPdo(); + $this->colorCli = new ColorCLI; + $this->_collectionsCleaning = new class + { + public function collectionsCleaner($subject, $groupName): array + { + return ['id' => 1, 'name' => 'COLL']; + } + }; + $this->xrefService = new XrefService; + $this->blacklistService = new BlacklistService; + $this->messageBuffer = 50000; + $this->_compressedHeaders = false; + $this->_partRepair = true; + $this->_newGroupScanByDays = false; + $this->_newGroupMessagesToScan = 50000; + $this->_newGroupDaysToScan = 3; + $this->_partRepairLimit = 15000; + $this->_partRepairMaxTries = 3; + $this->blackList = $this->whiteList = []; + } + + // Expose protected storeHeaders for direct testing. + public function publicStoreHeaders(array $headers): void + { + if (empty($this->groupMySQL)) { + $this->groupMySQL = ['id' => 1, 'name' => 'alt.test']; + } + $this->startCleaning = now(); + config(['tests.force_simulated_rollback' => false]); + $this->storeHeaders($headers); + } + + public function setAddToPartRepair(bool $val): void + { + $this->addToPartRepair = $val; + } + + // Simulate scan path minimally to test rollback + part repair queue logic without NNTP. + public function simulateScan(array $headers, array $group, bool $enablePartRepair = true): void + { + $this->groupMySQL = $group; + $this->first = $headers[0]['Number']; + $this->last = end($headers)['Number']; + $this->headersReceived = array_column($headers, 'Number'); + $this->addToPartRepair = $enablePartRepair; + $this->startCleaning = now(); + + // If we are simulating a failure, do not perform any inserts; just mark all as missed. + if ($this->failPartsInsert) { + if ($enablePartRepair) { + foreach (array_unique($this->headersReceived) as $num) { + DB::insert('INSERT INTO missed_parts (numberid, groups_id, attempts) VALUES (?, ?, 1)', [$num, $group['id']]); + } + } + return; + } + + // Normal path: process and insert. + $this->storeHeaders($headers); + + if ($enablePartRepair && ! empty($this->headersNotInserted)) { + foreach (array_unique($this->headersNotInserted) as $num) { + DB::insert('INSERT INTO missed_parts (numberid, groups_id, attempts) VALUES (?, ?, 1)', [$num, $group['id']]); + } + } + } + + // Force chunk failure to trigger rollback when flag set. + protected function flushPartsChunk(array $parts): bool + { + $this->flushCount++; + if ($this->failPartsInsert) { + if ($this->failAfterFlushCount === null) { + return false; // Always fail when flag set and no threshold provided. + } + if ($this->flushCount > $this->failAfterFlushCount) { + return false; // Fail after N successful flushes + } + } + + return parent::flushPartsChunk($parts); + } +} diff --git a/tests/TestCase.php b/tests/TestCase.php index fe1ffc2ff..87327ba36 100644 --- a/tests/TestCase.php +++ b/tests/TestCase.php @@ -6,5 +6,5 @@ use Illuminate\Foundation\Testing\TestCase as BaseTestCase; abstract class TestCase extends BaseTestCase { - // + use CreatesApplication; // Boot Laravel application for tests. }