Move TCP settlement to systemd worker

README.md

@@ -33,12 +33,14 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저
 - `monitor.php`: 모니터링 화면과 AJAX 응답
 - `common.php`: 외부 secret 로드와 공통 DB/API 함수
 - `collector_se.php`: 상태 수집 CLI/cron
+- `tcp_worker.php`: TCP 큐 처리와 늦은 사용량 정산 worker
+- `car-tcp-worker.service`: systemd service 예시
 - `sw.js`: service worker
 - `assets/`: 아이콘과 정적 자산
 
 ## 데이터/저장소
 
-- Car DB tables: 차량 상태, 로그, 차트용 기록
+- Car DB tables: 차량 상태, 로그, 차트용 기록, TCP 요청 큐
 - 사용량 수집값과 월별 보정 기준
 - `/home/seo/secret/car.php`: TCP, DB, token, 허용 IP 설정
 
@@ -48,13 +50,15 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저
 2. `monitor.php`는 상태 AJAX와 사용량 AJAX를 분리해 갱신합니다.
 3. 제어 요청은 token/IP 정책을 검증합니다.
 4. 명령 코드를 허용 목록과 대조합니다.
-5. TCP 게이트웨이로 전송하고 결과와 실패 reason을 반환합니다.
+5. 요청을 `car_tcp_request_queue`에 저장합니다.
+6. `car-tcp-worker.service`가 큐를 처리하며 TCP 게이트웨이로 전송합니다.
+7. API는 큐 상태를 4.95초까지만 기다린 뒤 결과 또는 timeout을 반환합니다.
 
 ## TCP timeout과 사용량 정산
 
 차량 명령 성공/실패 판단은 UI 안정성을 위해 `TCP_TOTAL_TIMEOUT` 기본값인 4.95초를 사용합니다. 이 시간 안에 정상 응답이 없으면 API는 기존처럼 실패를 반환하므로 자동화와 UI는 즉시 재시도 여부를 판단할 수 있습니다.
 
-단, 실제 모뎀이 4.95초 이후에 늦게 응답할 수 있으므로 사용량 기록은 별도로 정산합니다. 4.95초까지 정상 응답이 없고 송신이 완료된 요청은 `pcntl_fork()`로 자식 프로세스를 만들고, 같은 TCP 소켓을 최대 `TCP_USAGE_SETTLEMENT_TIMEOUT` 기본 60초까지 더 관찰합니다.
+단, 실제 모뎀이 4.95초 이후에 늦게 응답할 수 있으므로 사용량 기록은 별도로 정산합니다. API는 요청을 DB 큐에 넣고 4.95초까지만 기다립니다. `car-tcp-worker.service`는 해당 요청의 TCP 연결을 직접 열고, UI timeout 이후에도 같은 소켓을 최대 `TCP_USAGE_SETTLEMENT_TIMEOUT` 기본 60초까지 더 관찰합니다.
 
 - 60초 안에 정상 응답이 완성되면 `late_response_after_ui_timeout`으로 송신량과 수신량을 함께 기록합니다.
 - 60초까지 응답이 없으면 `late_total_timeout_60s_sent_only`로 송신 패킷만 사용량에 반영합니다.
@@ -63,12 +67,26 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저
 
 이 구조에서 `request_id`는 UI 응답과 늦은 사용량 정산 기록을 연결하는 기준입니다. 즉 명령 재시도 로직은 4.95초 기준을 유지하고, 데이터 사용량은 최대 60초까지 늦은 수신 패킷을 반영합니다.
 
+## systemd worker
+
+설치 파일은 `car-tcp-worker.service`입니다.
+
+```bash
+cp /var/www/seoul/car/car-tcp-worker.service /etc/systemd/system/car-tcp-worker.service
+systemctl daemon-reload
+systemctl enable --now car-tcp-worker.service
+systemctl status car-tcp-worker.service
+```
+
+큐 테이블은 코드에서 자동 생성합니다. Worker가 중지되면 API 요청은 큐에 쌓이지만 4.95초 안에 처리되지 않아 실패로 반환될 수 있습니다. 안전을 위해 worker가 뒤늦게 오래된 queued 요청을 발견하면 실제 TCP 송신 없이 `queue_expired_before_send`로 만료 처리합니다.
+
 ## 주요 함수/모듈
 
 - `common.php`: secret 로드와 공통 DB/API 함수
 - `api.php`: 상태 조회와 명령 전송
 - `monitor.php`: 화면 렌더링과 AJAX 응답
 - `collector_se.php`: 상태 수집과 중복 실행 방지
+- `tcp_worker.php`: TCP 요청 큐 처리와 60초 사용량 정산
 
 ## 보안
 
@@ -83,3 +101,4 @@ Car는 차량 모뎀/게이트웨이에서 받은 상태를 수집해 DB에 저
 - 명령별 rate limit과 감사 로그를 유지합니다.
 - 통신사 기준 데이터 사용량 보정값을 주기적으로 확인합니다.
 - 4.95초 timeout 직후에는 사용량 기록이 즉시 생기지 않을 수 있으며, 최대 60초 뒤 늦은 정산 행이 추가될 수 있습니다.
+- `systemctl status car-tcp-worker.service`와 `journalctl -u car-tcp-worker.service`로 worker 상태를 확인합니다.

car-tcp-worker.service

@@ -0,0 +1,16 @@
+[Unit]
+Description=Car TCP queue worker
+After=network-online.target mariadb.service mysql.service
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=/var/www/seoul/car
+ExecStart=/usr/bin/php /var/www/seoul/car/tcp_worker.php
+Restart=always
+RestartSec=2
+User=root
+Group=root
+
+[Install]
+WantedBy=multi-user.target

common.php

@@ -158,50 +158,300 @@ function record_tcp_usage(
     );
 }
 
-function finalize_late_tcp_usage(
-    $fp,
-    string $cmd,
-    string $usageSource,
-    ?string $requestId,
-    int $sentBytes,
-    string $initialResponse,
-    float $deadline
-): void {
-    ignore_user_abort(true);
-    @set_time_limit((int)ceil(max(5.0, TCP_USAGE_SETTLEMENT_TIMEOUT + 5.0)));
+function tcp_request_payload(string $cmd): string
+{
+    return json_encode([
+        'type' => 'R',
+        'type_sub' => 'car_controll',
+        'data' => [
+            'command' => '+SCMD=' . MODEM . '/C:' . $cmd,
+            'modem'   => MODEM,
+            'user'    => USER,
+            'uid'     => UID,
+            'type'    => TYPE,
+        ],
+    ], JSON_UNESCAPED_SLASHES);
+}
 
-    $response = $initialResponse;
-    $tcpError = '';
+function db_ensure_tcp_queue_schema(PDO $pdo): void
+{
+    static $done = false;
+    if ($done) {
+        return;
+    }
 
-    while (microtime(true) < $deadline) {
-        $read = [$fp];
-        $w = null;
-        $e = null;
+    $pdo->exec("CREATE TABLE IF NOT EXISTS car_tcp_request_queue (
+        id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
+        request_id VARCHAR(64) NOT NULL UNIQUE,
+        car_id VARCHAR(20) NOT NULL DEFAULT 'car',
+        created_at DATETIME NOT NULL,
+        updated_at DATETIME NOT NULL,
+        source VARCHAR(20) NOT NULL,
+        cmd VARCHAR(10) NOT NULL,
+        status VARCHAR(32) NOT NULL,
+        request_payload MEDIUMTEXT NOT NULL,
+        sent_bytes INT NOT NULL DEFAULT 0,
+        received_bytes INT NOT NULL DEFAULT 0,
+        total_bytes INT NOT NULL DEFAULT 0,
+        tcp_ok TINYINT NULL,
+        tcp_error VARCHAR(100) NULL,
+        raw_full MEDIUMTEXT NULL,
+        connect_ms INT NOT NULL DEFAULT 0,
+        read_ms INT NOT NULL DEFAULT 0,
+        tcp_ms INT NOT NULL DEFAULT 0,
+        ui_deadline_ts DOUBLE NOT NULL DEFAULT 0,
+        usage_deadline_ts DOUBLE NOT NULL DEFAULT 0,
+        worker_pid INT NULL,
+        KEY idx_status_id (status, id),
+        KEY idx_request_id (request_id)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
 
-        $remain = $deadline - microtime(true);
-        if ($remain <= 0) {
-            break;
+    $done = true;
+}
+
+function tcp_queue_now(): string
+{
+    return date('Y-m-d H:i:s');
+}
+
+function tcp_queue_update(PDO $pdo, string $requestId, array $values): void
+{
+    if (!$values) {
+        return;
+    }
+
+    $values['updated_at'] = tcp_queue_now();
+    $sets = [];
+    $params = [':request_id' => $requestId];
+
+    foreach ($values as $key => $value) {
+        $param = ':' . $key;
+        $sets[] = "$key = $param";
+        $params[$param] = $value;
+    }
+
+    $stmt = $pdo->prepare("UPDATE car_tcp_request_queue SET " . implode(', ', $sets) . " WHERE request_id = :request_id");
+    $stmt->execute($params);
+}
+
+function tcp_queue_insert(string $cmd, string $usageSource, string $requestId): array
+{
+    $pdo = db();
+    db_ensure_tcp_queue_schema($pdo);
+
+    $now = microtime(true);
+    $payload = tcp_request_payload($cmd);
+
+    $stmt = $pdo->prepare("INSERT INTO car_tcp_request_queue (
+        request_id, car_id, created_at, updated_at, source, cmd, status,
+        request_payload, ui_deadline_ts, usage_deadline_ts
+    ) VALUES (
+        :request_id, 'car', :created_at, :updated_at, :source, :cmd, 'queued',
+        :request_payload, :ui_deadline_ts, :usage_deadline_ts
+    )");
+    $stmt->execute([
+        ':request_id' => $requestId,
+        ':created_at' => tcp_queue_now(),
+        ':updated_at' => tcp_queue_now(),
+        ':source' => substr($usageSource, 0, 20),
+        ':cmd' => substr($cmd, 0, 10),
+        ':request_payload' => $payload,
+        ':ui_deadline_ts' => $now + TCP_TOTAL_TIMEOUT,
+        ':usage_deadline_ts' => $now + TCP_USAGE_SETTLEMENT_TIMEOUT,
+    ]);
+
+    return [
+        'request_id' => $requestId,
+        'cmd' => $cmd,
+        'source' => $usageSource,
+        'request_payload' => $payload,
+        'ui_deadline_ts' => $now + TCP_TOTAL_TIMEOUT,
+        'usage_deadline_ts' => $now + TCP_USAGE_SETTLEMENT_TIMEOUT,
+    ];
+}
+
+function tcp_queue_get(string $requestId): ?array
+{
+    $pdo = db();
+    db_ensure_tcp_queue_schema($pdo);
+
+    $stmt = $pdo->prepare("SELECT * FROM car_tcp_request_queue WHERE request_id = :request_id LIMIT 1");
+    $stmt->execute([':request_id' => $requestId]);
+    $row = $stmt->fetch();
+    return $row ?: null;
+}
+
+function tcp_queue_claim_next(PDO $pdo): ?array
+{
+    db_ensure_tcp_queue_schema($pdo);
+
+    $pdo->beginTransaction();
+    try {
+        $stmt = $pdo->query("SELECT * FROM car_tcp_request_queue WHERE status = 'queued' ORDER BY id ASC LIMIT 1 FOR UPDATE");
+        $row = $stmt->fetch();
+        if (!$row) {
+            $pdo->commit();
+            return null;
         }
 
-        $wait = min($remain, 1.0);
+        $update = $pdo->prepare("UPDATE car_tcp_request_queue SET status = 'processing', updated_at = :updated_at, worker_pid = :worker_pid WHERE request_id = :request_id");
+        $update->execute([
+            ':updated_at' => tcp_queue_now(),
+            ':worker_pid' => getmypid() ?: null,
+            ':request_id' => $row['request_id'],
+        ]);
+        $pdo->commit();
+
+        $row['status'] = 'processing';
+        $row['worker_pid'] = getmypid() ?: null;
+        return $row;
+    } catch (Throwable $e) {
+        if ($pdo->inTransaction()) {
+            $pdo->rollBack();
+        }
+        throw $e;
+    }
+}
+
+function tcp_queue_record_usage_and_status(
+    PDO $pdo,
+    array $job,
+    string $status,
+    int $sentBytes,
+    int $receivedBytes,
+    bool $tcpOk,
+    string $tcpError,
+    string $rawFull = '',
+    int $connectMs = 0,
+    int $readMs = 0
+): void {
+    $requestId = (string)$job['request_id'];
+    $source = (string)$job['source'];
+    $cmd = (string)$job['cmd'];
+
+    db_insert_data_usage($pdo, $source, $cmd, $sentBytes, $receivedBytes, $tcpOk, $tcpError, $requestId);
+
+    tcp_queue_update($pdo, $requestId, [
+        'status' => $status,
+        'sent_bytes' => max(0, $sentBytes),
+        'received_bytes' => max(0, $receivedBytes),
+        'total_bytes' => max(0, $sentBytes + $receivedBytes),
+        'tcp_ok' => $tcpOk ? 1 : 0,
+        'tcp_error' => $tcpError !== '' ? substr($tcpError, 0, 100) : null,
+        'raw_full' => $rawFull !== '' ? $rawFull : null,
+        'connect_ms' => max(0, $connectMs),
+        'read_ms' => max(0, $readMs),
+        'tcp_ms' => max(0, $connectMs + $readMs),
+    ]);
+}
+
+function tcp_queue_maybe_store_status(PDO $pdo, array $job, string $rawFull): void
+{
+    $trimError = '';
+    $rawTrim = make_trim($rawFull, $trimError);
+    if ($rawTrim === '') {
+        return;
+    }
+
+    $data = parse_trim($rawTrim);
+    if (!is_valid_status_data($data)) {
+        return;
+    }
+
+    db_insert_status($pdo, (string)$job['cmd'], $rawFull, $rawTrim, $data);
+}
+
+function tcp_worker_process_job(array $job): void
+{
+    $pdo = db(true);
+    db_ensure_tcp_queue_schema($pdo);
+
+    $cmd = (string)$job['cmd'];
+    $requestId = (string)$job['request_id'];
+    $payload = (string)$job['request_payload'];
+    $uiDeadline = (float)$job['ui_deadline_ts'];
+    $usageDeadline = (float)$job['usage_deadline_ts'];
+    $connectStart = microtime(true);
+    $sentBytes = 0;
+    $receivedBytes = 0;
+    $response = '';
+    $tcpError = '';
+    $uiTimedOut = false;
+
+    if (microtime(true) >= $uiDeadline) {
+        tcp_queue_record_usage_and_status($pdo, $job, 'expired_before_send', 0, 0, false, 'queue_expired_before_send', '', 0, 0);
+        return;
+    }
+
+    $connectTimeout = max(0.01, min(TCP_TOTAL_TIMEOUT, $uiDeadline - microtime(true)));
+    $fp = @stream_socket_client("tcp://" . TCP_HOST . ":" . TCP_PORT, $errno, $errstr, $connectTimeout);
+    $connectMs = (int)round((microtime(true) - $connectStart) * 1000);
+
+    if (!$fp) {
+        tcp_queue_record_usage_and_status($pdo, $job, 'connect_error', 0, 0, false, "connect_error: $errno $errstr", '', $connectMs, 0);
+        return;
+    }
+
+    stream_set_blocking($fp, false);
+    $written = @fwrite($fp, $payload);
+    if ($written === false || $written < strlen($payload)) {
+        $sentBytes = ($written === false) ? 0 : (int)$written;
+        @fclose($fp);
+        tcp_queue_record_usage_and_status($pdo, $job, 'write_failed', $sentBytes, 0, false, 'write_failed', '', $connectMs, 0);
+        return;
+    }
+
+    $sentBytes = (int)$written;
+    tcp_queue_update($pdo, $requestId, [
+        'sent_bytes' => $sentBytes,
+        'total_bytes' => $sentBytes,
+    ]);
+
+    $readStart = microtime(true);
+    while (microtime(true) < $usageDeadline) {
+        $now = microtime(true);
+
+        if (!$uiTimedOut && $now >= $uiDeadline) {
+            $uiTimedOut = true;
+            tcp_queue_update($pdo, $requestId, [
+                'status' => 'ui_timeout_pending',
+                'tcp_error' => 'total_timeout_4_95s_late_settlement_pending',
+                'received_bytes' => strlen($response),
+                'total_bytes' => $sentBytes + strlen($response),
+                'connect_ms' => $connectMs,
+                'read_ms' => (int)round(($now - $readStart) * 1000),
+                'tcp_ms' => $connectMs + (int)round(($now - $readStart) * 1000),
+            ]);
+        }
+
+        $nextWake = min($usageDeadline, $uiTimedOut ? $now + 1.0 : $uiDeadline);
+        $wait = max(0.001, min(1.0, $nextWake - $now));
         $sec = (int)$wait;
         $usec = (int)(($wait - $sec) * 1000000);
 
+        $read = [$fp];
+        $w = null;
+        $e = null;
         $sel = @stream_select($read, $w, $e, $sec, $usec);
         if ($sel === false) {
-            $tcpError = 'late_stream_select_failed';
+            $tcpError = 'stream_select_failed';
             break;
         }
 
         if ($sel > 0) {
             $chunk = @fread($fp, 4096);
             if ($chunk === false) {
-                $tcpError = 'late_read_failed';
+                $tcpError = 'read_failed';
                 break;
             }
 
             if ($chunk !== '') {
                 $response .= $chunk;
+                $receivedBytes = strlen($response);
+                tcp_queue_update($pdo, $requestId, [
+                    'received_bytes' => $receivedBytes,
+                    'total_bytes' => $sentBytes + $receivedBytes,
+                ]);
 
                 if (preg_match(RAW_FULL_REGEX, trim($response))) {
                     break;
@@ -216,77 +466,87 @@ function finalize_late_tcp_usage(
 
     @fclose($fp);
 
+    $readMs = (int)round((microtime(true) - $readStart) * 1000);
     $receivedBytes = strlen($response);
     $trimmed = trim($response);
 
     if ($trimmed === '') {
-        db_insert_data_usage(db(true), $usageSource, $cmd, $sentBytes, 0, false, 'late_total_timeout_60s_sent_only', $requestId);
+        tcp_queue_record_usage_and_status($pdo, $job, 'settled_sent_only', $sentBytes, 0, false, 'late_total_timeout_60s_sent_only', '', $connectMs, $readMs);
         return;
     }
 
     if (!preg_match(RAW_FULL_REGEX, $trimmed)) {
-        db_insert_data_usage(
-            db(true),
-            $usageSource,
-            $cmd,
+        tcp_queue_record_usage_and_status(
+            $pdo,
+            $job,
+            'settled_partial',
             $sentBytes,
             $receivedBytes,
             false,
             $tcpError !== '' ? $tcpError : 'late_invalid_or_partial_response',
-            $requestId
+            $trimmed,
+            $connectMs,
+            $readMs
         );
         return;
     }
 
-    db_insert_data_usage(db(true), $usageSource, $cmd, $sentBytes, $receivedBytes, true, 'late_response_after_ui_timeout', $requestId);
-
-    $trimError = '';
-    $rawTrim = make_trim($trimmed, $trimError);
-    if ($rawTrim === '') {
-        return;
-    }
-
-    $data = parse_trim($rawTrim);
-    if (!is_valid_status_data($data)) {
-        return;
-    }
-
-    try {
-        db_insert_status(db(true), $cmd, $trimmed, $rawTrim, $data);
-    } catch (Throwable $e) {
-        // 늦은 응답 상태 저장 실패가 사용량 정산을 되돌리면 안 된다.
-    }
+    $status = $uiTimedOut ? 'settled_late' : 'settled';
+    $error = $uiTimedOut ? 'late_response_after_ui_timeout' : '';
+    tcp_queue_record_usage_and_status($pdo, $job, $status, $sentBytes, $receivedBytes, true, $error, $trimmed, $connectMs, $readMs);
+    tcp_queue_maybe_store_status($pdo, $job, $trimmed);
 }
 
-function schedule_late_tcp_usage_settlement(
-    $fp,
-    string $cmd,
-    string $usageSource,
-    ?string $requestId,
-    int $sentBytes,
-    string $initialResponse = ''
-): bool {
-    if (!function_exists('pcntl_fork') || $sentBytes <= 0) {
-        return false;
+function tcp_request_wait_for_queue(
+    string $requestId,
+    int &$tcpMs,
+    int &$connectMs,
+    int &$readMs,
+    string &$tcpError,
+    int &$sentBytes,
+    int &$receivedBytes
+): string {
+    $deadline = microtime(true) + TCP_TOTAL_TIMEOUT;
+
+    while (microtime(true) < $deadline) {
+        $row = tcp_queue_get($requestId);
+        if (!$row) {
+            usleep(50000);
+            continue;
+        }
+
+        $status = (string)$row['status'];
+        $sentBytes = (int)($row['sent_bytes'] ?? 0);
+        $receivedBytes = (int)($row['received_bytes'] ?? 0);
+        $connectMs = (int)($row['connect_ms'] ?? 0);
+        $readMs = (int)($row['read_ms'] ?? 0);
+        $tcpMs = (int)($row['tcp_ms'] ?? ($connectMs + $readMs));
+        $tcpError = (string)($row['tcp_error'] ?? '');
+
+        if ($status === 'settled' && (int)($row['tcp_ok'] ?? 0) === 1) {
+            return (string)($row['raw_full'] ?? '');
+        }
+
+        if (in_array($status, ['ui_timeout_pending', 'connect_error', 'write_failed', 'settled_sent_only', 'settled_partial'], true)) {
+            if ($tcpError === '') {
+                $tcpError = $status;
+            }
+            return '';
+        }
+
+        usleep(50000);
     }
 
-    $deadline = microtime(true) + TCP_USAGE_SETTLEMENT_TIMEOUT;
-    $pid = @pcntl_fork();
-
-    if ($pid === -1) {
-        return false;
+    $row = tcp_queue_get($requestId);
+    if ($row) {
+        $sentBytes = (int)($row['sent_bytes'] ?? 0);
+        $receivedBytes = (int)($row['received_bytes'] ?? 0);
+        $connectMs = (int)($row['connect_ms'] ?? 0);
+        $readMs = (int)($row['read_ms'] ?? 0);
+        $tcpMs = (int)($row['tcp_ms'] ?? ($connectMs + $readMs));
     }
-
-    if ($pid > 0) {
-        return true;
-    }
-
-    if (function_exists('posix_setsid')) {
-        @posix_setsid();
-    }
-
-    finalize_late_tcp_usage($fp, $cmd, $usageSource, $requestId, $sentBytes, $initialResponse, $deadline);
-    exit(0);
+    $tcpError = 'queue_wait_timeout_late_settlement_pending';
+    return '';
 }
 
 function tcp_request(
@@ -308,133 +568,27 @@ function tcp_request(
     $sentBytes = 0;
     $receivedBytes = 0;
 
-    $request = json_encode([
-        'type' => 'R',
-        'type_sub' => 'car_controll',
-        'data' => [
-            'command' => '+SCMD=' . MODEM . '/C:' . $cmd,
-            'modem'   => MODEM,
-            'user'    => USER,
-            'uid'     => UID,
-            'type'    => TYPE,
-        ],
-    ], JSON_UNESCAPED_SLASHES);
-
-    $t0 = microtime(true);
-
-    $deadline = $t0 + TCP_TOTAL_TIMEOUT;
-
-    $remain = $deadline - microtime(true);
-    if ($remain <= 0) {
-        record_tcp_usage($usageSource, $cmd, 0, 0, false, 'deadline_expired_before_connect', $requestId);
-        return '';
+    if ($requestId === null || $requestId === '') {
+        $requestId = substr(bin2hex(random_bytes(16)), 0, 32);
     }
-
-    $fp = @stream_socket_client(
-        "tcp://" . TCP_HOST . ":" . TCP_PORT,
-        $errno,
-        $errstr,
-        $remain
-    );
-    $connectMs = (int)round((microtime(true) - $t0) * 1000);
-
-    if (!$fp) {
-        $tcpError = "connect_error: $errno $errstr";
-        $tcpMs = $connectMs;
+
+    try {
+        tcp_queue_insert($cmd, $usageSource, $requestId);
+    } catch (Throwable $e) {
+        $tcpError = 'queue_insert_failed';
         record_tcp_usage($usageSource, $cmd, 0, 0, false, $tcpError, $requestId);
         return '';
     }
 
-    stream_set_blocking($fp, false);
-
-    $written = @fwrite($fp, $request);
-    if ($written === false || $written < strlen($request)) {
-        $sentBytes = ($written === false) ? 0 : (int)$written;
-        fclose($fp);
-        $tcpError = 'write_failed';
-        record_tcp_usage($usageSource, $cmd, $sentBytes, 0, false, $tcpError, $requestId);
-        return '';
-    }
-    $sentBytes = (int)$written;
-
-    $r0 = microtime(true);
-    $response = '';
-    while (microtime(true) < $deadline) {
-        $read = [$fp];
-        $w = null;
-        $e = null;
-
-        $remain = $deadline - microtime(true);
-        if ($remain <= 0) break;
-
-        $sec = (int)$remain;
-        $usec = (int)(($remain - $sec) * 1000000);
-
-        $sel = @stream_select($read, $w, $e, $sec, $usec);
-        if ($sel === false) {
-            $tcpError = 'stream_select_failed';
-            break;
-        }
-
-        if ($sel > 0) {
-            $chunk = @fread($fp, 4096);
-            if ($chunk === false) {
-                $tcpError = 'read_failed';
-                break;
-            }
-
-            if ($chunk !== '') {
-                $response .= $chunk;
-
-                if (preg_match(RAW_FULL_REGEX, trim($response))) {
-                    break;
-                }
-            }
-        }
-
-        if (feof($fp)) {
-            break;
-        }
-    }
-
-    $readMs = (int)round((microtime(true) - $r0) * 1000);
-
-    $tcpMs = $connectMs + $readMs;
-    $receivedBytes = strlen($response);
-
-    if (trim($response) === '') {
-        if ($tcpError === '') {
-            $tcpError = 'total_timeout_4_95s';
-        }
-        $scheduled = $tcpError === 'total_timeout_4_95s'
-            && schedule_late_tcp_usage_settlement($fp, $cmd, $usageSource, $requestId, $sentBytes, $response);
-        if ($scheduled) {
-            fclose($fp);
-            $tcpError = 'total_timeout_4_95s_late_settlement_pending';
-        } else {
-            fclose($fp);
-            record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, false, $tcpError, $requestId);
-        }
-        return '';
-    }
-
-    if (!preg_match(RAW_FULL_REGEX, trim($response))) {
-        $tcpError = 'invalid_or_partial_response';
-        $scheduled = schedule_late_tcp_usage_settlement($fp, $cmd, $usageSource, $requestId, $sentBytes, $response);
-        if ($scheduled) {
-            fclose($fp);
-            $tcpError = 'invalid_or_partial_response_late_settlement_pending';
-        } else {
-            fclose($fp);
-            record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, false, $tcpError, $requestId);
-        }
-        return '';
-    }
-
-    fclose($fp);
-    record_tcp_usage($usageSource, $cmd, $sentBytes, $receivedBytes, true, $tcpError, $requestId);
-
-    return $response;
+    return tcp_request_wait_for_queue(
+        $requestId,
+        $tcpMs,
+        $connectMs,
+        $readMs,
+        $tcpError,
+        $sentBytes,
+        $receivedBytes
+    );
 }
 
 function make_trim(string $rawFull, string &$trimError = ''): string

tcp_worker.php

@@ -0,0 +1,44 @@
+<?php
+declare(strict_types=1);
+
+require __DIR__ . '/common.php';
+
+const WORKER_IDLE_SLEEP_USEC = 200000;
+
+if (PHP_SAPI !== 'cli') {
+    fwrite(STDERR, "tcp_worker.php must run from CLI\n");
+    exit(1);
+}
+
+$lockFp = fopen('/tmp/car_tcp_worker.lock', 'c');
+if ($lockFp === false) {
+    fwrite(STDERR, "failed to open worker lock\n");
+    exit(1);
+}
+
+if (!flock($lockFp, LOCK_EX | LOCK_NB)) {
+    fwrite(STDERR, "worker already running\n");
+    exit(0);
+}
+
+register_shutdown_function(static function () use ($lockFp): void {
+    @flock($lockFp, LOCK_UN);
+    @fclose($lockFp);
+});
+
+db_ensure_tcp_queue_schema(db());
+
+while (true) {
+    try {
+        $job = tcp_queue_claim_next(db());
+        if (!$job) {
+            usleep(WORKER_IDLE_SLEEP_USEC);
+            continue;
+        }
+
+        tcp_worker_process_job($job);
+    } catch (Throwable $e) {
+        fwrite(STDERR, date('c') . ' ' . $e->getMessage() . "\n");
+        usleep(1000000);
+    }
+}