Fix R15: reject out-of-order MQTT state updates (concurrency)

With multiple Horizon workers, ingest jobs for one entity can finish out of
order and overwrite newer state with older. The listener now stamps each message
with a µs receive time (observed_at); IngestShellyMessage applies state only when
the incoming message is newer (race-safe via a conditional update + unique guard),
and broadcasts only when applied. Added a feature test for the ordering guard.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/Console/Commands/MqttListenCommand.php

@@ -33,7 +33,9 @@ class MqttListenCommand extends Command
                 foreach (ShellyTopics::subscriptions() as $topic) {
                     $this->client->subscribe($topic, function (string $topic, string $message): void {
                         // H2: parse + dispatch only — never do DB/HTTP work in the loop.
-                        IngestShellyMessage::dispatch($topic, $message);
+                        // Stamp the receive time (µs) so out-of-order job completion can't
+                        // overwrite newer state with older (concurrent workers).
+                        IngestShellyMessage::dispatch($topic, $message, (int) round(microtime(true) * 1_000_000));
                     }, 0);
                 }
 

app/Jobs/IngestShellyMessage.php

@@ -5,14 +5,20 @@ namespace App\Jobs;
 use App\Events\DeviceStateChanged;
 use App\Models\Device;
 use App\Models\DeviceState;
+use App\Models\Entity;
 use App\Support\Mqtt\ShellyNormalizer;
 use App\Support\Mqtt\ShellyTopics;
 use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Database\UniqueConstraintViolationException;
 use Illuminate\Foundation\Queue\Queueable;
 
 /**
  * The single ingest path for Shelly status messages (H4). The MQTT callback only
  * dispatches this (H2); here we resolve the device, upsert current state and broadcast.
+ *
+ * `observedAt` is the µs receive time from the listener. Because concurrent Horizon
+ * workers can finish out of order, state is only written when this message is newer than
+ * what is stored — an older message can never overwrite a newer one.
  */
 class IngestShellyMessage implements ShouldQueue
 {
@@ -21,6 +27,7 @@ class IngestShellyMessage implements ShouldQueue
     public function __construct(
         public string $topic,
         public string $payload,
+        public ?int $observedAt = null,
     ) {}
 
     public function handle(): void
@@ -43,6 +50,8 @@ class IngestShellyMessage implements ShouldQueue
             return;
         }
 
+        $observedAt = $this->observedAt ?? (int) round(microtime(true) * 1_000_000);
+
         $device->forceFill(['last_seen_at' => now()])->save();
 
         foreach (ShellyNormalizer::normalize($component, $data) as $update) {
@@ -51,12 +60,46 @@ class IngestShellyMessage implements ShouldQueue
                 ['type' => $update['type']],
             );
 
-            DeviceState::updateOrCreate(
-                ['entity_id' => $entity->id],
-                ['state' => $update['state']],
-            );
-
-            DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
+            if ($this->applyState($entity->id, $update['state'], $observedAt)) {
+                DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
+            }
         }
     }
+
+    /**
+     * Monotonic, race-safe upsert. Returns true when this message's state was applied,
+     * false when it was stale (an equal-or-newer state is already stored).
+     */
+    private function applyState(int $entityId, array $state, int $observedAt): bool
+    {
+        $encoded = json_encode($state);
+
+        // Overwrite an existing row only when this message is newer.
+        $updated = DeviceState::query()
+            ->where('entity_id', $entityId)
+            ->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
+            ->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]);
+
+        if ($updated > 0) {
+            return true;
+        }
+
+        // No row yet → insert (unique(entity_id) guards against a concurrent insert).
+        if (! DeviceState::where('entity_id', $entityId)->exists()) {
+            try {
+                DeviceState::create(['entity_id' => $entityId, 'state' => $state, 'observed_at' => $observedAt]);
+
+                return true;
+            } catch (UniqueConstraintViolationException) {
+                // Lost the insert race — fall through and retry the conditional update.
+                return DeviceState::query()
+                    ->where('entity_id', $entityId)
+                    ->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
+                    ->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]) > 0;
+            }
+        }
+
+        // Existing row is equal-or-newer → this message is stale.
+        return false;
+    }
 }

app/Models/DeviceState.php

@@ -10,9 +10,9 @@ use Illuminate\Database\Eloquent\Relations\BelongsTo;
  */
 class DeviceState extends Model
 {
-    protected $fillable = ['entity_id', 'state'];
+    protected $fillable = ['entity_id', 'state', 'observed_at'];
 
-    protected $casts = ['state' => 'array'];
+    protected $casts = ['state' => 'array', 'observed_at' => 'integer'];
 
     public function entity(): BelongsTo
     {

database/migrations/2026_07_17_203001_add_observed_at_to_device_states.php

@@ -0,0 +1,24 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    public function up(): void
+    {
+        Schema::table('device_states', function (Blueprint $table) {
+            // Microseconds since epoch, stamped when the listener received the message.
+            // Used to reject out-of-order updates (concurrent Horizon workers).
+            $table->unsignedBigInteger('observed_at')->nullable()->after('state');
+        });
+    }
+
+    public function down(): void
+    {
+        Schema::table('device_states', function (Blueprint $table) {
+            $table->dropColumn('observed_at');
+        });
+    }
+};

tests/Feature/MqttTest.php

@@ -34,6 +34,22 @@ class MqttTest extends TestCase
         $this->assertTrue($device->refresh()->last_seen_at->gt(now()->subMinute()));
     }
 
+    public function test_out_of_order_messages_do_not_overwrite_newer_state(): void
+    {
+        $device = Device::create([
+            'name' => 'Lampe', 'vendor' => 'Shelly', 'protocol' => 'mqtt',
+            'config' => ['mqtt_prefix' => 'shelly-lampe'], 'status' => 'active',
+        ]);
+
+        // Newer message (higher observed_at) says on=true …
+        (new IngestShellyMessage('shelly-lampe/status/switch:0', json_encode(['output' => true]), 2_000))->handle();
+        // … an older one arriving late must NOT overwrite it.
+        (new IngestShellyMessage('shelly-lampe/status/switch:0', json_encode(['output' => false]), 1_000))->handle();
+
+        $switch = $device->entities()->where('key', 'switch:0')->first();
+        $this->assertTrue(data_get($switch->state->state, 'on'));
+    }
+
     public function test_ingest_ignores_unknown_device(): void
     {
         (new IngestShellyMessage('shelly-nope/status/switch:0', json_encode(['output' => true])))->handle();