fix(telematics): retention pruner no longer scans the whole device_events table every run

server/migrations/2026_10_10_000001_add_trashed_index_to_device_events.php

@@ -0,0 +1,40 @@
+<?php
+
+use Fleetbase\FleetOps\Support\Database\TableIndexes;
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+/**
+ * fleetops:prune-telematics-data purges soft-deleted rows on every run; without an
+ * index on deleted_at that sweep walks the whole table even when nothing is trashed.
+ */
+return new class extends Migration {
+    private const TABLE   = 'device_events';
+    private const INDEX   = 'device_events_company_deleted_at_index';
+    private const COLUMNS = ['company_uuid', 'deleted_at'];
+
+    public function up(): void
+    {
+        if (!Schema::hasTable(self::TABLE) || !Schema::hasColumns(self::TABLE, self::COLUMNS)) {
+            return;
+        }
+
+        if (TableIndexes::covers(self::TABLE, self::COLUMNS)) {
+            return;
+        }
+
+        Schema::table(self::TABLE, fn (Blueprint $table) => $table->index(self::COLUMNS, self::INDEX));
+    }
+
+    public function down(): void
+    {
+        if (!Schema::hasTable(self::TABLE)) {
+            return;
+        }
+
+        if ((TableIndexes::for(self::TABLE)[self::INDEX] ?? null) === self::COLUMNS) {
+            Schema::table(self::TABLE, fn (Blueprint $table) => $table->dropIndex(self::INDEX));
+        }
+    }
+};

server/migrations/2026_10_10_000002_add_trashed_index_to_positions.php

@@ -0,0 +1,40 @@
+<?php
+
+use Fleetbase\FleetOps\Support\Database\TableIndexes;
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+/**
+ * fleetops:prune-telematics-data purges soft-deleted rows on every run; without an
+ * index on deleted_at that sweep walks the whole table even when nothing is trashed.
+ */
+return new class extends Migration {
+    private const TABLE   = 'positions';
+    private const INDEX   = 'positions_company_deleted_at_index';
+    private const COLUMNS = ['company_uuid', 'deleted_at'];
+
+    public function up(): void
+    {
+        if (!Schema::hasTable(self::TABLE) || !Schema::hasColumns(self::TABLE, self::COLUMNS)) {
+            return;
+        }
+
+        if (TableIndexes::covers(self::TABLE, self::COLUMNS)) {
+            return;
+        }
+
+        Schema::table(self::TABLE, fn (Blueprint $table) => $table->index(self::COLUMNS, self::INDEX));
+    }
+
+    public function down(): void
+    {
+        if (!Schema::hasTable(self::TABLE)) {
+            return;
+        }
+
+        if ((TableIndexes::for(self::TABLE)[self::INDEX] ?? null) === self::COLUMNS) {
+            Schema::table(self::TABLE, fn (Blueprint $table) => $table->dropIndex(self::INDEX));
+        }
+    }
+};

server/src/Console/Commands/PruneTelematicsData.php

@@ -2,6 +2,7 @@
 
 namespace Fleetbase\FleetOps\Console\Commands;
 
+use Carbon\Carbon;
 use Fleetbase\FleetOps\Support\Telematics\Retention\RetentionPolicy;
 use Illuminate\Console\Command;
 use Illuminate\Support\Collection;
@@ -21,6 +22,9 @@ class PruneTelematicsData extends Command
 
     public const LOCK = 'fleetops:prune-telematics-data';
 
+    /** Cache key prefix for the per-scope compaction watermark (see pruneDeviceEvents). */
+    public const WATERMARK_PREFIX = 'fleetops:prune-telematics-data:compacted:';
+
     protected $signature = 'fleetops:prune-telematics-data
         {--company= : Only prune data for this company (uuid or public id)}
         {--orphans-only : Only prune data without an existing company or connection}
@@ -36,6 +40,9 @@ class PruneTelematicsData extends Command
     protected int $maxBatches = 50;
     protected bool $dryRun    = false;
 
+    /** table => company uuids present in that table whose company no longer exists; resolved once per run. */
+    protected array $orphanCompaniesByTable = [];
+
     public function handle(): int
     {
         $tables = $this->requestedTables();
@@ -142,32 +149,62 @@ protected function pruneCompany(string $companyUuid, RetentionPolicy $policy, ar
         }
         $inbox = $telematics ? fn ($query) => $query->whereIn('telematic_uuid', $telematics) : null;
 
-        return $this->prune($rows, $inbox, $policy, $tables);
+        return $this->prune($rows, $inbox, $policy, $tables, $companyUuid);
     }
 
     /**
      * Rows whose company or connection no longer exists fall back to the system defaults.
      */
     protected function pruneOrphans(RetentionPolicy $policy, array $tables): array
     {
-        $rows = fn ($query) => $query->where(function ($where) {
-            $where->whereNull('company_uuid')->orWhereNotIn('company_uuid', DB::table('companies')->select('uuid'));
-        });
+        $rows  = fn ($query) => $this->scopeOrphanRows($query);
         $inbox = fn ($query) => $query->whereNotExists(function ($connections) use ($query) {
             $connections->selectRaw('1')
                 ->from('telematics')
                 ->join('companies', 'companies.uuid', '=', 'telematics.company_uuid')
                 ->whereColumn('telematics.uuid', $query->from . '.telematic_uuid');
         });
 
-        return $this->prune($rows, $inbox, $policy, $tables);
+        return $this->prune($rows, $inbox, $policy, $tables, 'orphans');
+    }
+
+    /**
+     * Rows without a company, plus rows whose company uuid no longer exists. The gone
+     * companies are resolved once per table from the distinct company uuids it holds, so
+     * every batch ranges over the (company_uuid, created_at) index instead of evaluating
+     * a NOT IN subquery against every row.
+     */
+    protected function scopeOrphanRows($query)
+    {
+        $table = $query->from;
+        $gone  = $this->orphanCompaniesByTable[$table] ??= $this->orphanCompanies($table);
+
+        return $query->where(function ($where) use ($gone) {
+            $where->whereNull('company_uuid');
+            if ($gone !== []) {
+                $where->orWhereIn('company_uuid', $gone);
+            }
+        });
+    }
+
+    /**
+     * @return string[]
+     */
+    protected function orphanCompanies(string $table): array
+    {
+        return DB::table($table)
+            ->whereNotNull('company_uuid')
+            ->whereNotIn('company_uuid', DB::table('companies')->select('uuid'))
+            ->groupBy('company_uuid')
+            ->pluck('company_uuid')
+            ->all();
     }
 
-    protected function prune(\Closure $rows, ?\Closure $inbox, RetentionPolicy $policy, array $tables): array
+    protected function prune(\Closure $rows, ?\Closure $inbox, RetentionPolicy $policy, array $tables, string $scope): array
     {
         $stats = $this->emptyStats();
         if (in_array('device_events', $tables, true)) {
-            $stats['tables']['device_events'] = $this->pruneDeviceEvents($rows, $policy);
+            $stats['tables']['device_events'] = $this->pruneDeviceEvents($rows, $policy, $scope);
         }
         if (in_array('positions', $tables, true)) {
             $stats['tables']['positions'] = $this->prunePositions($rows, $policy);
@@ -188,40 +225,69 @@ protected function prune(\Closure $rows, ?\Closure $inbox, RetentionPolicy $poli
         return $stats;
     }
 
-    protected function pruneDeviceEvents(\Closure $rows, RetentionPolicy $policy): array
+    protected function pruneDeviceEvents(\Closure $rows, RetentionPolicy $policy, string $scope): array
     {
         $stats  = $this->emptyTableStats();
         $cutoff = $policy->cutoff('event_retention_days');
         // Soft-deleted events only grow the table; purge them regardless of age.
-        $this->deleteInBatches('device_events', 'id', fn ($query) => $rows($query)->whereNotNull('deleted_at'), $stats);
+        $this->deleteInBatches('device_events', 'id', fn ($query) => $rows($query)->whereNotNull('deleted_at'), $stats, 'deleted_at');
         if ($cutoff) {
-            $this->deleteInBatches('device_events', 'id', fn ($query) => $rows($query)->where('created_at', '<', $cutoff->toDateTimeString()), $stats);
+            $this->deleteInBatches('device_events', 'id', fn ($query) => $rows($query)->where('created_at', '<', $cutoff->toDateTimeString()), $stats, 'created_at');
         }
         if ($policy->compactsEvents() && ($compactAt = $policy->cutoff('event_compact_after_days'))) {
             // Keep the event (type, severity, location, normalized data) but drop the raw provider blobs.
             // Rows past the delete cutoff are left to the delete sweep rather than rewritten first.
-            $this->updateInBatches(
+            // Everything older than the watermark was compacted by an earlier complete sweep, so
+            // only the rows that aged into the window since then are visited; without it every
+            // run re-scanned the whole history looking for blobs that were already gone.
+            $watermark = $this->compactionWatermark($scope);
+            $since     = $cutoff && (!$watermark || $cutoff->greaterThan($watermark)) ? $cutoff : $watermark;
+            $completed = $this->updateInBatches(
                 'device_events',
                 'id',
-                function ($query) use ($rows, $cutoff, $compactAt) {
+                function ($query) use ($rows, $since, $compactAt) {
                     $query = $rows($query)->where('created_at', '<', $compactAt->toDateTimeString())->where(fn ($where) => $where->whereNotNull('payload')->orWhereNotNull('meta'));
 
-                    return $cutoff ? $query->where('created_at', '>=', $cutoff->toDateTimeString()) : $query;
+                    return $since ? $query->where('created_at', '>=', $since->toDateTimeString()) : $query;
                 },
                 ['payload' => null, 'meta' => null],
-                $stats
+                $stats,
+                'created_at'
             );
+            if ($completed && !$this->dryRun) {
+                $this->rememberCompactionWatermark($scope, $compactAt);
+            }
         }
 
         return $stats;
     }
 
+    protected function compactionWatermark(string $scope): ?Carbon
+    {
+        try {
+            $value = Cache::get(self::WATERMARK_PREFIX . $scope);
+        } catch (\Throwable) {
+            return null;
+        }
+
+        return $value ? Carbon::parse($value) : null;
+    }
+
+    protected function rememberCompactionWatermark(string $scope, Carbon $compactedBefore): void
+    {
+        try {
+            Cache::forever(self::WATERMARK_PREFIX . $scope, $compactedBefore->toDateTimeString());
+        } catch (\Throwable) {
+            // A cache outage only costs the next run a wider scan.
+        }
+    }
+
     protected function prunePositions(\Closure $rows, RetentionPolicy $policy): array
     {
         $stats = $this->emptyTableStats();
-        $this->deleteInBatches('positions', 'id', fn ($query) => $rows($query)->whereNotNull('deleted_at'), $stats);
+        $this->deleteInBatches('positions', 'id', fn ($query) => $rows($query)->whereNotNull('deleted_at'), $stats, 'deleted_at');
         if ($cutoff = $policy->cutoff('position_retention_days')) {
-            $this->deleteInBatches('positions', 'id', fn ($query) => $rows($query)->where('created_at', '<', $cutoff->toDateTimeString()), $stats);
+            $this->deleteInBatches('positions', 'id', fn ($query) => $rows($query)->where('created_at', '<', $cutoff->toDateTimeString()), $stats, 'created_at');
         }
 
         return $stats;
@@ -232,7 +298,7 @@ protected function pruneDeliveries(\Closure $inbox, RetentionPolicy $policy): ar
         $stats = $this->emptyTableStats();
         foreach (['processed' => 'processed_retention_hours', 'quarantined' => 'quarantine_retention_days'] as $status => $key) {
             if ($cutoff = $policy->cutoff($key)) {
-                $this->deleteInBatches('telematic_deliveries', 'uuid', fn ($query) => $inbox($query)->where('status', $status)->where('updated_at', '<', $cutoff->toDateTimeString()), $stats);
+                $this->deleteInBatches('telematic_deliveries', 'uuid', fn ($query) => $inbox($query)->where('status', $status)->where('updated_at', '<', $cutoff->toDateTimeString()), $stats, 'updated_at');
             }
         }
 
@@ -244,46 +310,55 @@ protected function pruneSyncRuns(\Closure $inbox, RetentionPolicy $policy): arra
         $stats = $this->emptyTableStats();
         if ($cutoff = $policy->cutoff('sync_run_retention_days')) {
             // In-flight runs are recovered by the drain command, never expired here.
-            $this->deleteInBatches('telematic_sync_runs', 'uuid', fn ($query) => $inbox($query)->whereNotIn('status', ['fetching', 'ingesting'])->where('updated_at', '<', $cutoff->toDateTimeString()), $stats);
+            $this->deleteInBatches('telematic_sync_runs', 'uuid', fn ($query) => $inbox($query)->whereNotIn('status', ['fetching', 'ingesting'])->where('updated_at', '<', $cutoff->toDateTimeString()), $stats, 'updated_at');
         }
 
         return $stats;
     }
 
-    protected function deleteInBatches(string $table, string $key, \Closure $scope, array &$stats): void
+    /**
+     * Batches are ordered by the column the scope ranges over (the retention indexes lead
+     * with company_uuid and that column), so each pick is an index range rather than a
+     * primary-key walk across the whole table. Returns true when the scope was drained.
+     */
+    protected function deleteInBatches(string $table, string $key, \Closure $scope, array &$stats, ?string $orderBy = null): bool
     {
         if ($this->dryRun) {
             $stats['deleted'] += $scope(DB::table($table))->count();
 
-            return;
+            return true;
         }
         for ($batch = 0; $batch < $this->maxBatches; $batch++) {
-            $ids = $scope(DB::table($table))->orderBy($key)->limit($this->batchSize)->pluck($key);
+            $ids = $scope(DB::table($table))->orderBy($orderBy ?? $key)->limit($this->batchSize)->pluck($key);
             if ($ids->isEmpty()) {
-                return;
+                return true;
             }
             $stats['batches']++;
             $stats['deleted'] += DB::table($table)->whereIn($key, $ids->all())->delete();
         }
         $stats['capped'] = true;
+
+        return false;
     }
 
-    protected function updateInBatches(string $table, string $key, \Closure $scope, array $values, array &$stats): void
+    protected function updateInBatches(string $table, string $key, \Closure $scope, array $values, array &$stats, ?string $orderBy = null): bool
     {
         if ($this->dryRun) {
             $stats['compacted'] += $scope(DB::table($table))->count();
 
-            return;
+            return true;
         }
         for ($batch = 0; $batch < $this->maxBatches; $batch++) {
-            $ids = $scope(DB::table($table))->orderBy($key)->limit($this->batchSize)->pluck($key);
+            $ids = $scope(DB::table($table))->orderBy($orderBy ?? $key)->limit($this->batchSize)->pluck($key);
             if ($ids->isEmpty()) {
-                return;
+                return true;
             }
             $stats['batches']++;
             $stats['compacted'] += DB::table($table)->whereIn($key, $ids->all())->update($values);
         }
         $stats['capped'] = true;
+
+        return false;
     }
 
     protected function report(string $label, array $stats): void