release: bump version to 2610.1

The purge ran one DELETE over the whole retention cutoff in a single
transaction, spiking undo/binlog and holding row locks on a table the
fetch cron writes to. Now it loops in batches of 5000 rows with a short
pause between batches, keeping each commit small.

README.md

@@ -225,7 +225,7 @@ One row per connection event ingested from any node.
 | `banner_len` | `INT UNSIGNED` | Length of the banner in bytes |
 | `created_at` | `DATETIME` | Row insertion time |
 
-**Indexes:** `(node_id, occurred_at)` for cursor lookups and per-node queries; `src_ip` and `dst_port` for the top-N aggregations.
+**Indexes:** `(node_id, occurred_at)` for cursor lookups and per-node queries; `occurred_at` for time-window queries, the recent-connections sort, and the purge; `(src_ip, occurred_at)` for the frequent-IP aggregation; `dst_port` for the top-port aggregation.
 
 ---
 
@@ -507,7 +507,7 @@ If you deployed before `last_event_at` was added to the schema, run `setup.php`
 php /var/www/portspoof_concentrator/setup.php
 ```
 
-It will add the missing column and drop the legacy `uq_event` unique key if present. No data is lost.
+It will add the missing column, drop the legacy `uq_event` unique key, and add the `occurred_at` / `(src_ip, occurred_at)` indexes if present. No data is lost.
 
 ---
 

includes/functions.php

@@ -75,19 +75,16 @@ function fetch_node_api(array $node, string $path): mixed {
  * Ingest connection events from portspoof_py that are strictly newer than
  * $last_event_at (ISO 8601 string, or null to ingest everything).
  *
+ * New rows are written with multi-row INSERTs in chunks, inside a single
+ * transaction per call, to keep the number of round trips and commits low.
+ *
  * Returns ['inserted' => int, 'new_max_ts' => string|null]
  * where new_max_ts is the DATETIME(6) of the newest row inserted, or null if
  * nothing new was inserted.
  */
 function ingest_connections(int $node_id, array $events, ?string $last_event_at): array {
-    $inserted   = 0;
     $new_max_ts = null;
-
-    $s = db()->prepare(
-        'INSERT INTO connections
-         (node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len)
-         VALUES (?, ?, ?, ?, ?, ?, ?)'
-    );
+    $rows       = [];
 
     foreach ($events as $ev) {
         $raw_ts = $ev['timestamp'] ?? null;
@@ -103,7 +100,7 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at)
             continue;
         }
 
-        $s->execute([
+        $rows[] = [
             $node_id,
             $ts,
             $ev['src_ip']   ?? '',
@@ -111,15 +108,40 @@ function ingest_connections(int $node_id, array $events, ?string $last_event_at)
             (int)($ev['dst_port']  ?? 0),
             $ev['banner_hex'] ?? null,
             (int)($ev['banner_len'] ?? 0),
-        ]);
-        $inserted++;
+        ];
 
         if ($new_max_ts === null || $ts > $new_max_ts) {
             $new_max_ts = $ts;
         }
     }
 
-    return ['inserted' => $inserted, 'new_max_ts' => $new_max_ts];
+    if ($rows === []) {
+        return ['inserted' => 0, 'new_max_ts' => null];
+    }
+
+    $pdo = db();
+    $pdo->beginTransaction();
+    try {
+        foreach (array_chunk($rows, 200) as $chunk) {
+            $values = implode(', ', array_fill(0, count($chunk), '(?, ?, ?, ?, ?, ?, ?)'));
+            $params = [];
+            foreach ($chunk as $r) {
+                $params = array_merge($params, $r);
+            }
+            $s = $pdo->prepare(
+                'INSERT INTO connections
+                 (node_id, occurred_at, src_ip, src_port, dst_port, banner_hex, banner_len)
+                 VALUES ' . $values
+            );
+            $s->execute($params);
+        }
+        $pdo->commit();
+    } catch (PDOException $e) {
+        $pdo->rollBack();
+        throw $e;
+    }
+
+    return ['inserted' => count($rows), 'new_max_ts' => $new_max_ts];
 }
 
 /**
@@ -236,15 +258,34 @@ function frequent_ips(int $min_connections, ?string $since = null): array {
 
 /**
  * Delete connections older than retention_days setting.
+ * Rows are removed in batches so each transaction stays small — a single
+ * giant DELETE would lock a large part of the table while the fetch cron
+ * keeps inserting rows.
+ *
  * Returns the number of rows deleted.
  */
 function purge_old_connections(): int {
-    $days = max(1, (int)get_setting('retention_days', '7'));
+    $days   = max(1, (int)get_setting('retention_days', '7'));
+    $cutoff = date('Y-m-d H:i:s', time() - $days * 86400);
+
     $s = db()->prepare(
-        'DELETE FROM connections WHERE occurred_at < NOW() - INTERVAL ? DAY'
+        'DELETE FROM connections
+         WHERE occurred_at < ?
+         ORDER BY id
+         LIMIT 5000'
     );
-    $s->execute([$days]);
-    return $s->rowCount();
+
+    $total = 0;
+    do {
+        $s->execute([$cutoff]);
+        $deleted = $s->rowCount();
+        $total += $deleted;
+        if ($deleted > 0) {
+            usleep(100_000); // 100 ms between batches to ease lock pressure
+        }
+    } while ($deleted >= 5000);
+
+    return $total;
 }
 
 // ── Upstream version check ────────────────────────────────────────────────────
@@ -252,18 +293,24 @@ function purge_old_connections(): int {
 define('UPSTREAM_VERSION_URL', 'https://git.ny.daprogs.com/api/v1/repos/DAProgs/portspoof_concentrator/raw/version.php?ref=main');
 define('UPSTREAM_VERSION_CACHE', __DIR__ . '/../upstream_version.cache');
 define('UPSTREAM_VERSION_TTL', 3600); // seconds
+define('UPSTREAM_VERSION_FAIL_TTL', 900); // seconds — how long to cache a failed check
 
 /**
  * Fetch the upstream version string from the git repo, with a 1-hour file cache.
  * Returns null silently on any failure so the page always loads.
+ *
+ * Failures are cached briefly too, so an unreachable git server does not
+ * add a cURL wait to every dashboard load (the page auto-refreshes every 30 s).
  */
 function fetch_upstream_version(): ?string {
-    // Return cached value if still fresh
+    // Return cached value (success or failure) if still fresh
     if (file_exists(UPSTREAM_VERSION_CACHE)) {
         $cached = json_decode(file_get_contents(UPSTREAM_VERSION_CACHE), true);
-        if (isset($cached['version'], $cached['checked_at'])
-            && (time() - $cached['checked_at']) < UPSTREAM_VERSION_TTL) {
-            return $cached['version'];
+        if (isset($cached['checked_at'])) {
+            $ttl = isset($cached['version']) ? UPSTREAM_VERSION_TTL : UPSTREAM_VERSION_FAIL_TTL;
+            if ((time() - $cached['checked_at']) < $ttl) {
+                return $cached['version'] ?? null;
+            }
         }
     }
 
@@ -271,27 +318,30 @@ function fetch_upstream_version(): ?string {
     curl_setopt_array($ch, [
         CURLOPT_RETURNTRANSFER => true,
         CURLOPT_TIMEOUT        => 5,
+        CURLOPT_CONNECTTIMEOUT => 2,
         CURLOPT_FOLLOWLOCATION => true,
     ]);
     $body = curl_exec($ch);
     $code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
     curl_close($ch);
 
-    if ($body === false || $code !== 200) {
-        return null;
+    if ($body !== false && $code === 200
+        && preg_match("/define\s*\(\s*'APP_VERSION'\s*,\s*'([^']+)'\s*\)/", $body, $m)) {
+        file_put_contents(
+            UPSTREAM_VERSION_CACHE,
+            json_encode(['version' => $m[1], 'checked_at' => time()]),
+            LOCK_EX
+        );
+        return $m[1];
     }
 
-    if (!preg_match("/define\s*\(\s*'APP_VERSION'\s*,\s*'([^']+)'\s*\)/", $body, $m)) {
-        return null;
-    }
-
-    $version = $m[1];
+    // Cache the failure so we don't retry on every page load
     file_put_contents(
         UPSTREAM_VERSION_CACHE,
-        json_encode(['version' => $version, 'checked_at' => time()]),
+        json_encode(['version' => null, 'checked_at' => time()]),
         LOCK_EX
     );
-    return $version;
+    return null;
 }
 
 /**
@@ -344,8 +394,6 @@ function global_stats(): array {
     $total = (int)$pdo->query('SELECT COUNT(*) FROM connections')->fetchColumn();
 
     $since = date('Y-m-d H:i:s', time() - RATE_WINDOW_SECONDS);
-    $recent = (int)$pdo->prepare('SELECT COUNT(*) FROM connections WHERE occurred_at >= ?')
-                        ->execute([$since]) ? : 0;
     $s = $pdo->prepare('SELECT COUNT(*) FROM connections WHERE occurred_at >= ?');
     $s->execute([$since]);
     $recent = (int)$s->fetchColumn();

schema.sql

@@ -27,7 +27,8 @@ CREATE TABLE IF NOT EXISTS connections (
     created_at  DATETIME        NOT NULL DEFAULT CURRENT_TIMESTAMP,
     PRIMARY KEY (id),
     KEY idx_node_occurred (node_id, occurred_at),
-    KEY idx_src_ip (src_ip),
+    KEY idx_occurred (occurred_at),
+    KEY idx_src_ip_occ (src_ip, occurred_at),
     KEY idx_dst_port (dst_port),
     CONSTRAINT fk_conn_node FOREIGN KEY (node_id) REFERENCES nodes (id) ON DELETE CASCADE
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

setup.php

@@ -41,6 +41,43 @@ if ($row > 0) {
     echo "Dropped legacy uq_event unique key.\n";
 }
 
+// Migration: add occurred_at index if upgrading from an older schema.
+// Time-window queries (dashboard, JSON API, purge) full-scanned the table
+// without it, so page loads slowed down as the table grew.
+$row = $pdo->query(
+    "SELECT COUNT(*) FROM information_schema.STATISTICS
+     WHERE TABLE_SCHEMA = DATABASE()
+       AND TABLE_NAME   = 'connections'
+       AND INDEX_NAME   = 'idx_occurred'"
+)->fetchColumn();
+if ($row == 0) {
+    $pdo->exec("ALTER TABLE connections ADD KEY idx_occurred (occurred_at)");
+    echo "Added idx_occurred index.\n";
+}
+
+// Migration: replace the standalone src_ip index with (src_ip, occurred_at)
+// so frequent_ips() resolves as a covering index scan.
+$row = $pdo->query(
+    "SELECT COUNT(*) FROM information_schema.STATISTICS
+     WHERE TABLE_SCHEMA = DATABASE()
+       AND TABLE_NAME   = 'connections'
+       AND INDEX_NAME   = 'idx_src_ip_occ'"
+)->fetchColumn();
+if ($row == 0) {
+    $pdo->exec("ALTER TABLE connections ADD KEY idx_src_ip_occ (src_ip, occurred_at)");
+    echo "Added idx_src_ip_occ index.\n";
+}
+$row = $pdo->query(
+    "SELECT COUNT(*) FROM information_schema.STATISTICS
+     WHERE TABLE_SCHEMA = DATABASE()
+       AND TABLE_NAME   = 'connections'
+       AND INDEX_NAME   = 'idx_src_ip'"
+)->fetchColumn();
+if ($row > 0) {
+    $pdo->exec("ALTER TABLE connections DROP INDEX idx_src_ip");
+    echo "Dropped redundant idx_src_ip index.\n";
+}
+
 // Migration: create settings table and seed default retention if upgrading
 $pdo->exec(
     "CREATE TABLE IF NOT EXISTS settings (

version.php

@@ -1,2 +1,2 @@
 <?php
-define('APP_VERSION', '2603.18');
+define('APP_VERSION', '2610.1');