feat(monitors): heartbeat token, ping recording and overdue verdict

server/internal/models/monitor.go

@@ -92,6 +92,8 @@ type Monitor struct {
 	// HeartbeatTokenHash is the SHA-256 of the ping token. The token itself is
 	// shown once, on create or rotate, and never stored.
 	HeartbeatTokenHash string `bson:"heartbeat_token_hash,omitempty" json:"-"`
+	// HeartbeatToken is the plaintext token, set only on the create response.
+	HeartbeatToken string `bson:"-" json:"heartbeat_token,omitempty"`
 }
 
 type Incident struct {

server/internal/services/heartbeats.go

@@ -0,0 +1,213 @@
+package services
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+const (
+	HeartbeatPing  = "ping"
+	HeartbeatStart = "start"
+	HeartbeatFail  = "fail"
+)
+
+// MaxHeartbeatBody bounds what a /fail request can put into an incident cause
+// and a notification. A job's stderr can be megabytes; the first kilobyte is
+// what a human reads.
+const MaxHeartbeatBody = 1024
+
+const defaultHeartbeatGraceSec = 300
+
+var ErrHeartbeatNotFound = errors.New("heartbeat not found")
+
+// ErrInvalidMonitor marks a validation failure, which handlers answer with 400.
+var ErrInvalidMonitor = errors.New("invalid monitor")
+
+func NewHeartbeatToken() (string, string, error) {
+	raw := make([]byte, 24)
+	if _, err := rand.Read(raw); err != nil {
+		return "", "", err
+	}
+	tok := base64.RawURLEncoding.EncodeToString(raw)
+	return tok, HashHeartbeatToken(tok), nil
+}
+
+func HashHeartbeatToken(token string) string {
+	sum := sha256.Sum256([]byte(token))
+	return hex.EncodeToString(sum[:])
+}
+
+func validateHeartbeat(t *models.MonitorTarget) error {
+	if t.PeriodSec < 60 {
+		return fmt.Errorf("%w: period_sec must be at least 60", ErrInvalidMonitor)
+	}
+	if t.GraceSec < 0 {
+		return fmt.Errorf("%w: grace_sec must not be negative", ErrInvalidMonitor)
+	}
+	if t.GraceSec == 0 {
+		t.GraceSec = defaultHeartbeatGraceSec
+	}
+	return nil
+}
+
+func failMessage(body string) string {
+	body = strings.TrimSpace(body)
+	if len(body) > MaxHeartbeatBody {
+		body = body[:MaxHeartbeatBody]
+	}
+	if body == "" {
+		return "reported failure"
+	}
+	return "reported failure: " + body
+}
+
+// heartbeatVerdict decides whether a heartbeat is overdue. A heartbeat that has
+// never pinged is never down: the clock starts at the first ping, so creating
+// one before the job is deployed does not page anyone.
+func heartbeatVerdict(m models.Monitor, now time.Time) (bool, string) {
+	grace := time.Duration(m.Target.GraceSec) * time.Second
+	if m.State.StartedAt != nil && now.After(m.State.StartedAt.Add(grace)) {
+		return true, fmt.Sprintf("started %s, never finished", m.State.StartedAt.UTC().Format(time.RFC3339))
+	}
+	if m.State.LastPingAt == nil {
+		return false, ""
+	}
+	deadline := m.State.LastPingAt.Add(time.Duration(m.Target.PeriodSec)*time.Second + grace)
+	if now.After(deadline) {
+		return true, fmt.Sprintf("no ping since %s", m.State.LastPingAt.UTC().Format(time.RFC3339))
+	}
+	return false, ""
+}
+
+// RecordHeartbeat applies one ping. The state change is a single
+// FindOneAndUpdate on the token hash, so two pings racing each other cannot
+// both read the old state and lose one of the writes.
+func RecordHeartbeat(token, kind, body string, now time.Time) error {
+	ctx, cancel := monCtx()
+	defer cancel()
+
+	filter := bson.M{"heartbeat_token_hash": HashHeartbeatToken(token), "type": models.MonitorHeartbeat, "enabled": true}
+	set := bson.M{"state.last_check_at": now}
+	unset := bson.M{}
+	var next, message string
+	switch kind {
+	case HeartbeatStart:
+		set["state.started_at"] = now
+	case HeartbeatPing:
+		set["state.last_ping_at"] = now
+		set["state.status"] = models.StatusUp
+		set["state.message"] = ""
+		unset["state.started_at"] = ""
+		next = models.StatusUp
+	case HeartbeatFail:
+		message = failMessage(body)
+		set["state.last_ping_at"] = now
+		set["state.status"] = models.StatusDown
+		set["state.message"] = message
+		unset["state.started_at"] = ""
+		next = models.StatusDown
+	default:
+		return fmt.Errorf("unknown heartbeat kind %q", kind)
+	}
+	upd := bson.M{"$set": set}
+	if len(unset) > 0 {
+		upd["$unset"] = unset
+	}
+
+	// ReturnDocument Before: the previous status and started_at are what the
+	// transition and the duration need.
+	var before models.Monitor
+	err := db.Col("monitors").FindOneAndUpdate(ctx, filter, upd,
+		options.FindOneAndUpdate().SetReturnDocument(options.Before)).Decode(&before)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return ErrHeartbeatNotFound
+	}
+	if err != nil {
+		return err
+	}
+	if kind == HeartbeatStart {
+		return nil
+	}
+
+	latency := 0
+	if kind == HeartbeatPing && before.State.StartedAt != nil {
+		latency = int(now.Sub(*before.State.StartedAt).Milliseconds())
+	}
+	if latency > 0 {
+		db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": before.MonitorID}, bson.M{"$set": bson.M{"state.latency_ms": latency}})
+	}
+	recordSample(ctx, &before, kind == HeartbeatPing, latency, now)
+	applyTransition(ctx, &before, "", "", before.State.Status, next, message, now)
+	return nil
+}
+
+// SweepHeartbeats marks overdue heartbeats down. Recovery only ever comes from
+// a ping, so the sweep never moves anything up.
+func SweepHeartbeats(now time.Time) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	cur, err := db.Col("monitors").Find(ctx, bson.M{
+		"type": models.MonitorHeartbeat, "enabled": true,
+		"state.status": bson.M{"$ne": models.StatusDown},
+	})
+	if err != nil {
+		log.Printf("heartbeats: list: %v", err)
+		return
+	}
+	var monitors []models.Monitor
+	if err := cur.All(ctx, &monitors); err != nil {
+		log.Printf("heartbeats: decode: %v", err)
+		return
+	}
+	for i := range monitors {
+		m := &monitors[i]
+		down, msg := heartbeatVerdict(*m, now)
+		if !down {
+			continue
+		}
+		// Conditional on status so a ping landing between the read and this
+		// write is not overwritten by a stale "down".
+		res, err := db.Col("monitors").UpdateOne(ctx,
+			bson.M{"monitor_id": m.MonitorID, "state.status": m.State.Status},
+			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 {
+			continue
+		}
+		recordSample(ctx, m, false, 0, now)
+		applyTransition(ctx, m, "", "", m.State.Status, models.StatusDown, msg, now)
+	}
+}
+
+func RotateHeartbeatToken(instanceID, monitorID string) (string, error) {
+	ctx, cancel := monCtx()
+	defer cancel()
+	tok, hash, err := NewHeartbeatToken()
+	if err != nil {
+		return "", err
+	}
+	res, err := db.Col("monitors").UpdateOne(ctx,
+		bson.M{"monitor_id": monitorID, "instance_id": instanceID, "type": models.MonitorHeartbeat},
+		bson.M{"$set": bson.M{"heartbeat_token_hash": hash}})
+	if err != nil {
+		return "", err
+	}
+	if res.MatchedCount == 0 {
+		return "", ErrHeartbeatNotFound
+	}
+	return tok, nil
+}

server/internal/services/heartbeats_test.go

@@ -0,0 +1,82 @@
+package services
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+)
+
+func hbMonitor(status string, lastPing, started *time.Time) models.Monitor {
+	return models.Monitor{
+		Type:   models.MonitorHeartbeat,
+		Target: models.MonitorTarget{PeriodSec: 3600, GraceSec: 300},
+		State:  models.MonitorState{Status: status, LastPingAt: lastPing, StartedAt: started},
+	}
+}
+
+func TestHeartbeatVerdict(t *testing.T) {
+	now := time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC)
+	at := func(d time.Duration) *time.Time { v := now.Add(-d); return &v }
+
+	cases := []struct {
+		name     string
+		m        models.Monitor
+		wantDown bool
+		wantMsg  string
+	}{
+		{"never pinged stays pending", hbMonitor(models.StatusPending, nil, nil), false, ""},
+		{"within period", hbMonitor(models.StatusUp, at(30*time.Minute), nil), false, ""},
+		{"inside grace", hbMonitor(models.StatusUp, at(62*time.Minute), nil), false, ""},
+		{"overdue", hbMonitor(models.StatusUp, at(66*time.Minute), nil), true, "no ping since"},
+		{"started inside grace", hbMonitor(models.StatusUp, at(10*time.Minute), at(4*time.Minute)), false, ""},
+		{"started never finished", hbMonitor(models.StatusUp, at(10*time.Minute), at(6*time.Minute)), true, "never finished"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			down, msg := heartbeatVerdict(c.m, now)
+			if down != c.wantDown || !strings.Contains(msg, c.wantMsg) {
+				t.Fatalf("got (%v,%q), want (%v, contains %q)", down, msg, c.wantDown, c.wantMsg)
+			}
+		})
+	}
+}
+
+func TestNewHeartbeatTokenHashes(t *testing.T) {
+	tok, hash, err := NewHeartbeatToken()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(tok) < 32 {
+		t.Fatalf("token too short: %d", len(tok))
+	}
+	if hash != HashHeartbeatToken(tok) || hash == tok {
+		t.Fatal("hash must be the SHA-256 of the token and differ from it")
+	}
+}
+
+func TestValidateHeartbeat(t *testing.T) {
+	tg := models.MonitorTarget{PeriodSec: 59}
+	if err := validateHeartbeat(&tg); err == nil {
+		t.Fatal("period under 60 must be rejected")
+	}
+	tg = models.MonitorTarget{PeriodSec: 60}
+	if err := validateHeartbeat(&tg); err != nil || tg.GraceSec != 300 {
+		t.Fatalf("grace should default to 300, got %d err %v", tg.GraceSec, err)
+	}
+	tg = models.MonitorTarget{PeriodSec: 60, GraceSec: -1}
+	if err := validateHeartbeat(&tg); err == nil {
+		t.Fatal("negative grace must be rejected")
+	}
+}
+
+func TestTruncateHeartbeatBody(t *testing.T) {
+	long := strings.Repeat("x", MaxHeartbeatBody+50)
+	if got := failMessage(long); len(got) != len("reported failure: ")+MaxHeartbeatBody {
+		t.Fatalf("len = %d", len(got))
+	}
+	if got := failMessage("  "); got != "reported failure" {
+		t.Fatalf("empty body message = %q", got)
+	}
+}

server/internal/services/monitors.go

@@ -188,6 +188,20 @@ func CreateMonitor(instanceID string, m *models.Monitor, tokenScope map[string]s
 	if m.Runner == "" {
 		m.Runner = models.RunnerServer
 	}
+	if m.Type == models.MonitorHeartbeat {
+		if err := validateHeartbeat(&m.Target); err != nil {
+			return nil, err
+		}
+		tok, hash, err := NewHeartbeatToken()
+		if err != nil {
+			return nil, err
+		}
+		m.HeartbeatTokenHash = hash
+		m.HeartbeatToken = tok
+		// A heartbeat is never run, so neither where nor how often applies.
+		m.Runner = models.RunnerServer
+		m.IntervalSec = 0
+	}
 	m.State = models.MonitorState{Status: models.StatusPending}
 	if _, err := db.Col("monitors").InsertOne(ctx, m); err != nil {
 		return nil, err
@@ -199,6 +213,33 @@ func UpdateMonitor(instanceID, monitorID string, upd bson.M, tokenScope map[stri
 	ctx, cancel := monCtx()
 	defer cancel()
 
+	existing, err := GetMonitor(instanceID, monitorID)
+	if err != nil {
+		return err
+	}
+	if existing == nil {
+		return fmt.Errorf("monitor not found")
+	}
+	if raw, present := upd["type"]; present {
+		if t, _ := raw.(string); t != existing.Type && (models.IsPassiveMonitor(t) || models.IsPassiveMonitor(existing.Type)) {
+			bad := existing.Type
+			if models.IsPassiveMonitor(t) {
+				bad = t
+			}
+			return fmt.Errorf("type cannot be changed to or from %s", bad)
+		}
+	}
+	if raw, present := upd["target"]; present && existing.Type == models.MonitorHeartbeat {
+		tg, ok := raw.(models.MonitorTarget)
+		if !ok {
+			return fmt.Errorf("target must be an object")
+		}
+		if err := validateHeartbeat(&tg); err != nil {
+			return err
+		}
+		upd["target"] = tg
+	}
+
 	if raw, present := upd["channel_ids"]; present {
 		ids, ok := raw.([]string)
 		if !ok {
@@ -232,7 +273,7 @@ func UpdateMonitor(instanceID, monitorID string, upd bson.M, tokenScope map[stri
 			upd["runner"] = models.RunnerServer
 		}
 	}
-	_, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}, bson.M{"$set": upd})
+	_, err = db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID, "instance_id": instanceID}, bson.M{"$set": upd})
 	return err
 }
 
@@ -359,35 +400,35 @@ func ingestResult(instanceID, runner, monitorID string, res checker.Result) erro
 		return err
 	}
 
-	bucket := now.Truncate(time.Hour)
-	up := 0
-	if res.Up {
-		up = 1
-	}
-
-	/* The sample is the same result at full resolution, expiring by TTL. It is
-	   written next to the rollup rather than instead of it: the rollup is what
-	   survives, the sample is what the sub-hour views read. */
-	db.Col("monitor_samples").InsertOne(ctx, models.MonitorSample{
-		InstanceID: m.InstanceID,
-		MonitorID:  monitorID,
-		At:         now,
-		Up:         res.Up,
-		LatencyMs:  res.LatencyMs,
-	})
-
-	db.Col("monitor_rollups").UpdateOne(ctx,
-		bson.M{"monitor_id": monitorID, "period_start": bucket},
-		bson.M{
-			"$inc":         bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)},
-			"$setOnInsert": bson.M{"instance_id": m.InstanceID},
-		},
-		options.UpdateOne().SetUpsert(true))
+	recordSample(ctx, m, res.Up, res.LatencyMs, now)
 
 	applyTransition(ctx, m, "", "", prev, newStatus, res.Message, now)
 	return nil
 }
 
+// recordSample writes one result at full resolution and folds it into the
+// hourly rollup, the two records every history view reads.
+func recordSample(ctx context.Context, m *models.Monitor, up bool, latencyMs int, now time.Time) {
+	u := 0
+	if up {
+		u = 1
+	}
+	db.Col("monitor_samples").InsertOne(ctx, models.MonitorSample{
+		InstanceID: m.InstanceID,
+		MonitorID:  m.MonitorID,
+		At:         now,
+		Up:         up,
+		LatencyMs:  latencyMs,
+	})
+	db.Col("monitor_rollups").UpdateOne(ctx,
+		bson.M{"monitor_id": m.MonitorID, "period_start": now.Truncate(time.Hour)},
+		bson.M{
+			"$inc":         bson.M{"checks": 1, "up_count": u, "sum_latency": int64(latencyMs)},
+			"$setOnInsert": bson.M{"instance_id": m.InstanceID},
+		},
+		options.UpdateOne().SetUpsert(true))
+}
+
 func notifyTransition(m *models.Monitor, serverName, oldStatus, newStatus, message string) {
 	if len(m.ChannelIDs) == 0 {
 		return