get()) { $this->warn('Ein anderer Aggregationslauf ist noch aktiv – dieser Aufruf tut nichts.'); return self::SUCCESS; } try { return $this->aggregate(); } finally { $lock->release(); } } private function aggregate(): int { $chunk = max(100, (int) $this->option('chunk')); $budget = min(self::LockSeconds - 60, max(0, (int) $this->option('budget'))); $startedAt = microtime(true); // Ungecacht (Audit N21): Die Marke ist Betriebszustand, kein // Admin-Schalter. Ein veralteter Cachewert ließe jeden Lauf an der // Prüfung unter der Sperre scheitern — dauerhaft. $watermark = AppSetting::getValueFresh(self::WatermarkKey); // Fehlende Marke heißt: noch nie initialisiert. Sie jetzt als 0 zu // lesen wäre fatal — `press_releases.hits` enthält den Bestand bereits, // ein Lauf ab 0 würde jedes Alt-Ereignis ein zweites Mal aufaddieren. // Das kann im Deploy-Fenster passieren, wenn der Scheduler den neuen // Task startet, bevor die Migration gelaufen ist. Also setzen wir die // Marke auf den Ist-Stand und zählen erst ab dem nächsten Aufruf. if ($watermark === null) { $latest = (int) PressReleaseEvent::query()->max('id'); AppSetting::setValue(self::WatermarkKey, $latest); $this->info("Wasserstandsmarke erstmalig auf Event-ID {$latest} gesetzt — der Bestand steckt bereits in `hits`."); return self::SUCCESS; } $watermark = (int) $watermark; $totalEvents = 0; $totalReleases = 0; while (true) { if (microtime(true) - $startedAt > $budget) { $this->warn(sprintf('Zeitbudget von %d s erreicht – der Rest folgt im nächsten Lauf.', $budget)); break; } // Obergrenze des Durchgangs zuerst festlegen, damit gleichzeitig // eintreffende Aufrufe die Summe nicht gegen die Marke verschieben. // Bewusst über die IDs statt über max(): eine Aggregatfunktion // ignoriert das limit und lieferte die Marke des GESAMTEN Rests. $ids = PressReleaseEvent::query() ->where('id', '>', $watermark) ->orderBy('id') ->limit($chunk) ->pluck('id'); if ($ids->isEmpty()) { break; } $upperBound = (int) $ids->last(); $counts = PressReleaseEvent::query() ->where('id', '>', $watermark) ->where('id', '<=', $upperBound) ->where('type', PressReleaseEventType::View->value) ->selectRaw('press_release_id, count(*) as views') ->groupBy('press_release_id') ->pluck('views', 'press_release_id'); // Zähler und Marke gehören in EINE Transaktion. Ohne sie wäre der // Lauf nur zwischen den Durchgängen wiederanlaufsicher: Bricht der // Prozess mitten im Chunk ab (OOM, Deploy, kill), sind ein Teil der // Zähler schon erhöht, die Marke steht aber noch auf dem alten // Stand — der nächste Lauf verbuchte genau diese Aufrufe ein // zweites Mal. try { $this->bookChunk($counts, $watermark, $upperBound, $totalEvents, $totalReleases); } catch (RuntimeException $exception) { report($exception); $this->error($exception->getMessage()); return self::FAILURE; } $watermark = $upperBound; } $this->info("Aufrufe verbucht: {$totalEvents} auf {$totalReleases} Pressemitteilungen (Stand Event-ID {$watermark})."); return self::SUCCESS; } /** * Zähler und Marke in EINER Transaktion; siehe Kommentar im Aufrufer. * * @param Collection $counts */ private function bookChunk(Collection $counts, int $watermark, int $upperBound, int &$totalEvents, int &$totalReleases): void { DB::transaction(function () use ($counts, $watermark, $upperBound, &$totalEvents, &$totalReleases): void { // Marke unter Zeilensperre gegen den erwarteten Stand prüfen: // hat ein anderer Lauf sie inzwischen verschoben (Lock // verfallen), würde dieser Durchgang dieselben Ereignisse ein // zweites Mal verbuchen (Audit U13). $current = AppSetting::query()->whereKey(self::WatermarkKey)->lockForUpdate()->first(); if ($current === null || (int) $current->value !== $watermark) { // Sollte je ein Cache-Leser die Marke veraltet halten: hier // verwerfen, damit der nächste Lauf sauber neu beginnt. AppSetting::forgetCached(self::WatermarkKey); throw new RuntimeException(sprintf( 'Wasserstandsmarke steht auf %s statt %d – ein anderer Lauf hat übernommen, Durchgang verworfen.', $current === null ? 'null' : (string) $current->value, $watermark, )); } foreach ($counts as $pressReleaseId => $views) { PressRelease::query() ->withoutGlobalScope(PortalScope::class) ->whereKey((int) $pressReleaseId) ->increment('hits', (int) $views); $totalReleases++; $totalEvents += (int) $views; } AppSetting::setValue(self::WatermarkKey, $upperBound); }); } }