fix(monitors): guard heartbeat sweep against a concurrent ping

server/internal/services/heartbeats.go

@@ -179,10 +179,13 @@ func SweepHeartbeats(now time.Time) {
 		if !down {
 			continue
 		}
-		// Conditional on status so a ping landing between the read and this
-		// write is not overwritten by a stale "down".
+		// Guarded on the exact fields the verdict was computed from, so a
+		// ping landing between the read and this write (which changes
+		// last_ping_at or started_at without necessarily changing status)
+		// makes the filter match nothing rather than overwriting a monitor
+		// that is no longer overdue.
 		res, err := db.Col("monitors").UpdateOne(ctx,
-			bson.M{"monitor_id": m.MonitorID, "state.status": m.State.Status},
+			sweepGuardFilter(*m),
 			bson.M{"$set": bson.M{"state.status": models.StatusDown, "state.message": msg, "state.last_check_at": now},
 				"$unset": bson.M{"state.started_at": ""}})
 		if err != nil || res.ModifiedCount == 0 {
@@ -193,6 +196,27 @@ func SweepHeartbeats(now time.Time) {
 	}
 }
 
+// sweepGuardFilter is the optimistic-concurrency filter for SweepHeartbeats'
+// update: it pins monitor_id, status, last_ping_at and started_at to the
+// values the verdict was computed from, so the update only applies when
+// nothing about the ping state changed underneath the sweep. last_ping_at is
+// absent before the first ping and started_at is $unset on every ping, so an
+// unset field is pinned with $exists:false rather than equality to nil.
+func sweepGuardFilter(m models.Monitor) bson.M {
+	filter := bson.M{"monitor_id": m.MonitorID, "state.status": m.State.Status}
+	if m.State.LastPingAt != nil {
+		filter["state.last_ping_at"] = *m.State.LastPingAt
+	} else {
+		filter["state.last_ping_at"] = bson.M{"$exists": false}
+	}
+	if m.State.StartedAt != nil {
+		filter["state.started_at"] = *m.State.StartedAt
+	} else {
+		filter["state.started_at"] = bson.M{"$exists": false}
+	}
+	return filter
+}
+
 func RotateHeartbeatToken(instanceID, monitorID string) (string, error) {
 	ctx, cancel := monCtx()
 	defer cancel()

server/internal/services/heartbeats_test.go

@@ -1,11 +1,13 @@
 package services
 
 import (
+	"reflect"
 	"strings"
 	"testing"
 	"time"
 
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+	"go.mongodb.org/mongo-driver/v2/bson"
 )
 
 func hbMonitor(status string, lastPing, started *time.Time) models.Monitor {
@@ -71,6 +73,39 @@ func TestValidateHeartbeat(t *testing.T) {
 	}
 }
 
+func TestSweepGuardFilterPinsFields(t *testing.T) {
+	now := time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC)
+
+	// Both set: filter pins status, last_ping_at and started_at to their
+	// exact values.
+	m := hbMonitor(models.StatusUp, &now, &now)
+	m.MonitorID = "mon-1"
+	got := sweepGuardFilter(m)
+	want := bson.M{
+		"monitor_id":         "mon-1",
+		"state.status":       models.StatusUp,
+		"state.last_ping_at": now,
+		"state.started_at":   now,
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("got %+v, want %+v", got, want)
+	}
+
+	// Both unset: filter requires the fields to still be absent.
+	m2 := hbMonitor(models.StatusPending, nil, nil)
+	m2.MonitorID = "mon-2"
+	got2 := sweepGuardFilter(m2)
+	want2 := bson.M{
+		"monitor_id":         "mon-2",
+		"state.status":       models.StatusPending,
+		"state.last_ping_at": bson.M{"$exists": false},
+		"state.started_at":   bson.M{"$exists": false},
+	}
+	if !reflect.DeepEqual(got2, want2) {
+		t.Fatalf("got %+v, want %+v", got2, want2)
+	}
+}
+
 func TestTruncateHeartbeatBody(t *testing.T) {
 	long := strings.Repeat("x", MaxHeartbeatBody+50)
 	if got := failMessage(long); len(got) != len("reported failure: ")+MaxHeartbeatBody {