Stream car usage aggregation

README.md

@@ -117,5 +117,6 @@ Worker 교체나 예외 종료로 `processing`, `ui_timeout_pending` 상태가
 - 명령별 rate limit과 감사 로그를 유지합니다.
 - 통신사 기준 데이터 사용량 보정값을 주기적으로 확인합니다. 2026-06-26 01:10 T world 실측 102.4MB(104,834KB) 기준 2026-06 보정값은 통신사 누적 107,350,016 bytes, 로컬 보정 전 누적 54,104,011 bytes, 배율 1.984142입니다.
 - 직전 배율 1.984648 기준 현재값은 107,377,429 bytes로 계산되어 T world보다 27,413 bytes(약 0.026MB) 높았고, 새 배율은 직전보다 0.000507 낮아졌습니다.
+- 데이터 사용량 집계는 성공 응답 평균을 SQL로 먼저 구하고, 월간/일간 행은 PHP에서 한 줄씩 스트리밍 처리합니다. 기존 보정 조건은 유지하면서 `fetchAll()` 메모리 사용을 제거해 PHP-FPM `memory_limit=128M`에서도 동작하도록 합니다.
 - 4.95초 timeout 직후에는 사용량 기록이 즉시 생기지 않을 수 있으며, 최대 60초 뒤 늦은 정산 행이 추가될 수 있습니다.
 - `systemctl status car-tcp-worker.service`와 `journalctl -u car-tcp-worker.service`로 worker 상태를 확인합니다.

monitor.php

@@ -204,24 +204,10 @@ function monthly_data_usage(PDO $pdo): array
     $monthStart = $now->modify('first day of this month')->setTime(0, 0, 0);
     $nextMonthStart = $monthStart->modify('first day of next month');
 
-    $stmt = $pdo->prepare(""
-        . "SELECT ts, source, cmd, sent_bytes, received_bytes, total_bytes, tcp_ok, tcp_error "
-        . "FROM car_data_usage "
-        . "WHERE id='car' AND ts >= :month_start AND ts < :next_month_start "
-        . "ORDER BY ts ASC"
-    );
-    $stmt->execute([
-        ':month_start' => $monthStart->format('Y-m-d H:i:s'),
-        ':next_month_start' => $nextMonthStart->format('Y-m-d H:i:s'),
-    ]);
-    $usageRows = $stmt->fetchAll();
-    $usage = calibrated_usage_rows($usageRows);
     $todayStart = $now->setTime(0, 0, 0);
     $tomorrowStart = $todayStart->modify('+1 day');
-    $todayUsage = calibrated_usage_rows(array_values(array_filter($usageRows, function (array $row) use ($todayStart, $tomorrowStart): bool {
-        $ts = strtotime((string)($row['ts'] ?? ''));
-        return $ts !== false && $ts >= $todayStart->getTimestamp() && $ts < $tomorrowStart->getTimestamp();
-    })));
+    $usage = calibrated_usage_range($pdo, $monthStart, $nextMonthStart);
+    $todayUsage = calibrated_usage_range($pdo, $todayStart, $tomorrowStart);
 
     $sentBytes = $usage['sent_bytes'];
     $receivedBytes = $usage['received_bytes'];
@@ -339,51 +325,57 @@ function floor_krw_10(float $value): int
     return (int)floor(max(0.0, $value) / 10) * 10;
 }
 
-function calibrated_usage_rows(array $rows): array
+function usage_success_averages(PDO $pdo, DateTimeInterface $start, DateTimeInterface $end): array
 {
-    $sentBytes = 0;
-    $receivedBytes = 0;
-    $totalBytes = 0;
     $successfulTotalBytes = 0;
     $successfulCount = 0;
-    $successByCmd = [];
-    $firstUsageTs = null;
-    $lastUsageTs = null;
+    $avgByCmd = [];
 
-    foreach ($rows as $row) {
+    $stmt = $pdo->prepare(""
+        . "SELECT cmd, SUM(total_bytes) AS total_bytes, COUNT(*) AS sample_count "
+        . "FROM car_data_usage "
+        . "WHERE id='car' AND ts >= :start_ts AND ts < :end_ts "
+        . "AND tcp_ok = 1 AND total_bytes > 0 "
+        . "GROUP BY cmd"
+    );
+    $stmt->execute([
+        ':start_ts' => $start->format('Y-m-d H:i:s'),
+        ':end_ts' => $end->format('Y-m-d H:i:s'),
+    ]);
+
+    while ($row = $stmt->fetch()) {
         $cmd = (string)($row['cmd'] ?? '');
-        $sent = (int)($row['sent_bytes'] ?? 0);
-        $received = (int)($row['received_bytes'] ?? 0);
         $total = (int)($row['total_bytes'] ?? 0);
-        $tcpOk = (int)($row['tcp_ok'] ?? 0) === 1;
-
-        $sentBytes += $sent;
-        $receivedBytes += $received;
-        $totalBytes += $total;
-
-        if ($firstUsageTs === null && !empty($row['ts'])) {
-            $firstUsageTs = (string)$row['ts'];
-        }
-        if (!empty($row['ts'])) {
-            $lastUsageTs = (string)$row['ts'];
+        $count = (int)($row['sample_count'] ?? 0);
+        if ($count <= 0) {
+            continue;
         }
 
-        if ($tcpOk && $total > 0) {
-            $successfulTotalBytes += $total;
-            $successfulCount++;
-            if (!isset($successByCmd[$cmd])) {
-                $successByCmd[$cmd] = ['total' => 0, 'count' => 0];
-            }
-            $successByCmd[$cmd]['total'] += $total;
-            $successByCmd[$cmd]['count']++;
-        }
+        $successfulTotalBytes += $total;
+        $successfulCount += $count;
+        $avgByCmd[$cmd] = (int)round($total / $count);
     }
 
     $globalAverage = $successfulCount > 0 ? (int)round($successfulTotalBytes / $successfulCount) : 0;
-    $avgByCmd = [];
-    foreach ($successByCmd as $cmd => $stat) {
-        $avgByCmd[$cmd] = $stat['count'] > 0 ? (int)round($stat['total'] / $stat['count']) : $globalAverage;
-    }
+
+    return [
+        'avg_by_cmd' => $avgByCmd,
+        'global_average' => $globalAverage,
+        'success_count' => $successfulCount,
+    ];
+}
+
+function calibrated_usage_range(PDO $pdo, DateTimeInterface $start, DateTimeInterface $end): array
+{
+    $averages = usage_success_averages($pdo, $start, $end);
+    $avgByCmd = $averages['avg_by_cmd'];
+    $globalAverage = $averages['global_average'];
+
+    $sentBytes = 0;
+    $receivedBytes = 0;
+    $totalBytes = 0;
+    $firstUsageTs = null;
+    $lastUsageTs = null;
 
     $adjustedTotalBytes = 0;
     $adjustedFailureBytes = 0;
@@ -391,47 +383,83 @@ function calibrated_usage_rows(array $rows): array
     $connectFailureCount = 0;
     $receiveGapExcludedCount = 0;
     $lastSuccessfulTs = null;
+    $sampleCount = 0;
 
-    foreach ($rows as $row) {
-        $cmd = (string)($row['cmd'] ?? '');
-        $sent = (int)($row['sent_bytes'] ?? 0);
-        $total = (int)($row['total_bytes'] ?? 0);
-        $tcpOk = (int)($row['tcp_ok'] ?? 0) === 1;
-        $rowTs = strtotime((string)($row['ts'] ?? ''));
+    $stmt = null;
+    $previousBuffered = $pdo->getAttribute(PDO::MYSQL_ATTR_USE_BUFFERED_QUERY);
+    $pdo->setAttribute(PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, false);
 
-        if ($tcpOk) {
-            $adjustedTotalBytes += $total;
-            if ($rowTs !== false) {
-                $lastSuccessfulTs = $rowTs;
+    try {
+        $stmt = $pdo->prepare(""
+            . "SELECT ts, cmd, sent_bytes, received_bytes, total_bytes, tcp_ok "
+            . "FROM car_data_usage "
+            . "WHERE id='car' AND ts >= :start_ts AND ts < :end_ts "
+            . "ORDER BY ts ASC"
+        );
+        $stmt->execute([
+            ':start_ts' => $start->format('Y-m-d H:i:s'),
+            ':end_ts' => $end->format('Y-m-d H:i:s'),
+        ]);
+
+        while ($row = $stmt->fetch()) {
+            $sampleCount++;
+            $cmd = (string)($row['cmd'] ?? '');
+            $sent = (int)($row['sent_bytes'] ?? 0);
+            $received = (int)($row['received_bytes'] ?? 0);
+            $total = (int)($row['total_bytes'] ?? 0);
+            $tcpOk = (int)($row['tcp_ok'] ?? 0) === 1;
+            $rowTs = strtotime((string)($row['ts'] ?? ''));
+
+            $sentBytes += $sent;
+            $receivedBytes += $received;
+            $totalBytes += $total;
+
+            if ($firstUsageTs === null && !empty($row['ts'])) {
+                $firstUsageTs = (string)$row['ts'];
             }
-            continue;
+            if (!empty($row['ts'])) {
+                $lastUsageTs = (string)$row['ts'];
+            }
+
+            if ($tcpOk) {
+                $adjustedTotalBytes += $total;
+                if ($rowTs !== false) {
+                    $lastSuccessfulTs = $rowTs;
+                }
+                continue;
+            }
+
+            if ($sent <= 0) {
+                // TCP 연결 자체가 실패한 건은 모뎀까지 명령이 전달되지 않은 것으로 보고 과금 추정에서 제외한다.
+                $connectFailureCount++;
+                continue;
+            }
+
+            $connectedFailureCount++;
+
+            if (
+                $lastSuccessfulTs === null ||
+                $rowTs === false ||
+                ($rowTs - $lastSuccessfulTs) > RECEIVE_GAP_LIMIT_SEC
+            ) {
+                $receiveGapExcludedCount++;
+                continue;
+            }
+
+            $estimatedBytes = $avgByCmd[$cmd] ?? $globalAverage;
+
+            if ($estimatedBytes <= 0) {
+                $estimatedBytes = max($sent, $total);
+            }
+
+            $adjustedTotalBytes += $estimatedBytes;
+            $adjustedFailureBytes += $estimatedBytes;
         }
-
-        if ($sent <= 0) {
-            // TCP 연결 자체가 실패한 건은 모뎀까지 명령이 전달되지 않은 것으로 보고 과금 추정에서 제외한다.
-            $connectFailureCount++;
-            continue;
+    } finally {
+        if ($stmt instanceof PDOStatement) {
+            $stmt->closeCursor();
         }
-
-        $connectedFailureCount++;
-
-        if (
-            $lastSuccessfulTs === null ||
-            $rowTs === false ||
-            ($rowTs - $lastSuccessfulTs) > RECEIVE_GAP_LIMIT_SEC
-        ) {
-            $receiveGapExcludedCount++;
-            continue;
-        }
-
-        $estimatedBytes = $avgByCmd[$cmd] ?? $globalAverage;
-
-        if ($estimatedBytes <= 0) {
-            $estimatedBytes = max($sent, $total);
-        }
-
-        $adjustedTotalBytes += $estimatedBytes;
-        $adjustedFailureBytes += $estimatedBytes;
+        $pdo->setAttribute(PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, $previousBuffered);
     }
 
     return [
@@ -443,8 +471,8 @@ function calibrated_usage_rows(array $rows): array
         'connected_failure_count' => $connectedFailureCount,
         'connect_failure_count' => $connectFailureCount,
         'receive_gap_excluded_count' => $receiveGapExcludedCount,
-        'sample_count' => count($rows),
-        'success_count' => $successfulCount,
+        'sample_count' => $sampleCount,
+        'success_count' => $averages['success_count'],
         'success_average_bytes' => $globalAverage,
         'first_usage_ts' => $firstUsageTs,
         'last_usage_ts' => $lastUsageTs,