Merge pull request 'Page the next person when nobody answers' (#16) from escalation into main
CI / chart (push) Successful in 1s
CI / test (push) Successful in 6s
CI / security (push) Successful in 12s

Reviewed-on: #16
This commit was merged in pull request #16.
This commit is contained in:
2026-09-20 16:43:03 +00:00
10 changed files with 1091 additions and 4 deletions
+46
View File
@@ -377,6 +377,50 @@ exhausts its retries. Written from the result rather than at enqueue, so the
timeline says what actually happened — and a page that never landed is visible timeline says what actually happened — and a page that never landed is visible
instead of looking the same as one that did. instead of looking the same as one that did.
### Escalation
Without a ladder, an unacknowledged incident re-pages the same topic every
`notify_repeat` forever. That is a louder version of the same silence: if the
person on call is asleep, out of signal, or has left, nothing else happens.
A team can configure an ordered ladder instead. Each level has a timeout and a
set of targets, and a target is either a named person or **whoever the team's
rota says is on call today** — the target that keeps working when the rota
changes and nobody remembers to edit the policy.
```
level 1 5m oncall the rota gets first refusal
level 2 5m user:bob then a named second
then repeat_count more rounds
then the team's fallback topic, once
```
When a level's timeout passes with the incident still `triggered`, the next
level is paged. Off the end of the ladder the whole thing runs again
`repeat_count` times, and after that the team's `fallback_topic` is paged once
as the end of the line. The incident stays open throughout: running out of
people to wake is not the same as somebody answering.
**Acknowledging or resolving stops it**, which is the point — continuing to wake
people after somebody has said "I have this" is how a tool teaches people to
mute it. **Snoozing pauses it**: a deliberate "not now" holds the ladder where
it is, and it resumes when the snooze runs out.
Every step is on the incident's timeline with the level and the names it woke,
so somebody reading it afterwards can tell why their phone rang at 04:00. A
level whose targets are all unreachable — no ntfy topic, a disabled account, an
empty rota — is recorded as `nobody reachable` and the ladder moves on rather
than stalling on a rung that cannot ring.
**Reminders and escalation never both run.** A team with a ladder gets
escalation; a team without keeps the reminder behaviour exactly as it was. Two
pages for one silence is the surest way to get a tool muted.
The ladder's `fallback_topic` is per team, unlike `TERDUT_NTFY_FALLBACK_TOPIC`,
which is the install-wide topic used when an incident opens with nobody on call.
They answer different questions: one is "nobody was scheduled", the other is
"everybody scheduled has been tried".
### Stale alert expiry ### Stale alert expiry
A resolved webhook is the only signal that an alert has stopped firing, so a A resolved webhook is the only signal that an alert has stopped firing, so a
@@ -578,6 +622,8 @@ and was removed in v0.13.0 once senders had moved onto keys.
| `GET` | `/api/teams/{teamID}/integrations` | member | List integrations. Never returns keys | | `GET` | `/api/teams/{teamID}/integrations` | member | List integrations. Never returns keys |
| `POST` | `/api/teams/{teamID}/integrations` | **owner** | Mint an integration `{"name","kind"}` — key and URL shown once | | `POST` | `/api/teams/{teamID}/integrations` | **owner** | Mint an integration `{"name","kind"}` — key and URL shown once |
| `DELETE` | `/api/teams/{teamID}/integrations/{integrationID}` | **owner** | Revoke an integration | | `DELETE` | `/api/teams/{teamID}/integrations/{integrationID}` | **owner** | Revoke an integration |
| `GET` | `/api/teams/{teamID}/escalation` | member | The team's [escalation ladder](#escalation) `{repeat_count, fallback_topic, levels[]}`. Empty levels means the team has none |
| `PUT` | `/api/teams/{teamID}/escalation` | **owner** | Replace it wholesale. `400` for a level with no targets or no timeout — a rung that pages nobody is a silence with a number on it |
| `GET` | `/api/teams/{teamID}/deadman` | member | The team's [dead man's switch](#dead-mans-switch) configuration `{matchers, timeout_seconds, severity}` | | `GET` | `/api/teams/{teamID}/deadman` | member | The team's [dead man's switch](#dead-mans-switch) configuration `{matchers, timeout_seconds, severity}` |
| `PUT` | `/api/teams/{teamID}/deadman` | **owner** | Replace it. `400` when no matcher names an `alertname`, because a switch that silently watches nothing is the failure this feature exists to prevent | | `PUT` | `/api/teams/{teamID}/deadman` | **owner** | Replace it. `400` when no matcher names an `alertname`, because a switch that silently watches nothing is the failure this feature exists to prevent |
+9 -3
View File
@@ -403,12 +403,18 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID in
} }
} }
// Queue the page, but do not send it here: this runs inside a transaction on // Queue the page, but do not send it here: this runs inside the webhook's
// a single-connection pool, so an HTTP call would hold up every other // transaction, and an HTTP call would hold a connection open across a
// request. The notifier picks the row up within a tick. // network round trip. The notifier picks the row up within a tick.
if err := enqueueOpened(ctx, q, notify, id, onCall); err != nil { if err := enqueueOpened(ctx, q, notify, id, onCall); err != nil {
return 0, err return 0, err
} }
// And start the escalation clock, if the team keeps one. In the same
// transaction, so an incident is never briefly open with nobody counting.
if err := startEscalation(ctx, q, id, teamID); err != nil {
return 0, err
}
return id, nil return id, nil
} }
+506
View File
@@ -0,0 +1,506 @@
package api
import (
"context"
"database/sql"
"log"
"net/http"
"strconv"
"strings"
"time"
)
// evEscalated records a rung of the ladder on the incident's timeline: which
// level, and who it woke.
const evEscalated = "escalated"
// escalationPolicy is a team's ladder, loaded whole. It is small — a handful of
// levels with a few targets each — and every use needs all of it, so there is
// no point reading it a level at a time.
type escalationPolicy struct {
teamID int64
repeatCount int64
fallbackTopic string
levels []escalationLevel
}
type escalationLevel struct {
id int64
position int64
timeout time.Duration
targets []escalationTarget
}
type escalationTarget struct {
kind string // "user" or "oncall"
userID *int64
}
// configured reports whether this team has anything to escalate through. A
// policy row with no levels is the same as no policy: the team gets the
// pre-escalation behaviour, which is reminders on the assignee's topic.
func (p *escalationPolicy) configured() bool { return p != nil && len(p.levels) > 0 }
// level returns the level at a 1-based position.
func (p *escalationPolicy) level(pos int64) (escalationLevel, bool) {
for _, l := range p.levels {
if l.position == pos {
return l, true
}
}
return escalationLevel{}, false
}
// loadEscalationPolicy reads one team's ladder. A team with no policy row
// returns nil, which every caller treats as "not configured" rather than as an
// error: most teams will never set one up.
func loadEscalationPolicy(ctx context.Context, q querier, teamID int64) (*escalationPolicy, error) {
p := &escalationPolicy{teamID: teamID}
err := q.QueryRowContext(ctx,
"SELECT repeat_count, fallback_topic FROM escalation_policies WHERE team_id = $1",
teamID).Scan(&p.repeatCount, &p.fallbackTopic)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
rows, err := q.QueryContext(ctx, `
SELECT l.id, l.position, l.timeout_seconds, t.kind, t.user_id
FROM escalation_levels l
LEFT JOIN escalation_targets t ON t.level_id = l.id
WHERE l.team_id = $1
ORDER BY l.position, t.id`, teamID)
if err != nil {
return nil, err
}
defer rows.Close()
byPosition := map[int64]int{} // position -> index in p.levels
for rows.Next() {
var id, position, timeout int64
var kind *string
var userID *int64
if err := rows.Scan(&id, &position, &timeout, &kind, &userID); err != nil {
return nil, err
}
idx, seen := byPosition[position]
if !seen {
p.levels = append(p.levels, escalationLevel{
id: id,
position: position,
timeout: time.Duration(timeout) * time.Second,
})
idx = len(p.levels) - 1
byPosition[position] = idx
}
// LEFT JOIN: a level with no targets yet still produces a row, with a
// NULL kind. It is a rung that pages nobody, which the API refuses to
// store but an older row could still hold.
if kind != nil {
p.levels[idx].targets = append(p.levels[idx].targets,
escalationTarget{kind: *kind, userID: userID})
}
}
return p, rows.Err()
}
// escalate advances every incident whose current level has run out of time.
//
// Runs on the notifier's tick, beside the reminder pass, because it is the same
// question asked differently: reminders ask "has this been ignored long
// enough to say it again", escalation asks "long enough to say it to somebody
// else". Sharing the tick means one query cadence and one outbox.
func escalate(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
rows, err := db.QueryContext(ctx, `
SELECT i.id, i.team_id, i.escalation_level, i.escalation_level_at, i.escalation_round
FROM incidents i
JOIN escalation_policies p ON p.team_id = i.team_id
WHERE i.resolved_at IS NULL
AND i.archived_at IS NULL
AND i.status = 'triggered'
AND (i.snoozed_until IS NULL OR i.snoozed_until <= $1)
AND i.escalation_level > 0`, time.Now().Unix())
if err != nil {
log.Printf("escalation: find due: %v", err)
return
}
type pending struct {
incidentID, teamID, level, round int64
levelAt int64
}
var due []pending
for rows.Next() {
var p pending
var levelAt *int64
if err := rows.Scan(&p.incidentID, &p.teamID, &p.level, &levelAt, &p.round); err != nil {
rows.Close()
log.Printf("escalation: scan: %v", err)
return
}
if levelAt == nil {
continue
}
p.levelAt = *levelAt
due = append(due, p)
}
rows.Close()
if err := rows.Err(); err != nil {
log.Printf("escalation: iterate: %v", err)
return
}
now := time.Now()
for _, d := range due {
policy, err := loadEscalationPolicy(ctx, db, d.teamID)
if err != nil {
log.Printf("escalation: load policy for team %d: %v", d.teamID, err)
continue
}
if !policy.configured() {
continue
}
current, ok := policy.level(d.level)
if !ok {
continue
}
if now.Sub(time.Unix(d.levelAt, 0)) < current.timeout {
continue
}
if err := advanceEscalation(ctx, db, cfg, policy, d.incidentID, d.level, d.round, now); err != nil {
log.Printf("escalation: advance incident %d: %v", d.incidentID, err)
}
}
}
// advanceEscalation moves one incident to its next rung, or off the end of the
// ladder.
//
// The whole move is one transaction: the level, the page and the timeline entry
// are one event, and an incident recorded as being at level 3 that nobody at
// level 3 was told about is the worst of the possible half-states.
func advanceEscalation(ctx context.Context, db *sql.DB, cfg NotifyConfig, policy *escalationPolicy, incidentID, level, round int64, now time.Time) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback() //nolint:errcheck
next := level + 1
nextRound := round
if _, ok := policy.level(next); !ok {
// Off the end. Either start the chain again, or make the last call.
if round < policy.repeatCount {
next, nextRound = 1, round+1
} else {
if err := escalationExhausted(ctx, tx, policy, incidentID, now); err != nil {
return err
}
return tx.Commit()
}
}
target, ok := policy.level(next)
if !ok {
return nil
}
paged, err := pageLevel(ctx, tx, cfg, policy, incidentID, target)
if err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
UPDATE incidents
SET escalation_level = $1, escalation_level_at = $2, escalation_round = $3
WHERE id = $4`, next, now.Unix(), nextRound, incidentID); err != nil {
return err
}
detail := "level " + strconv.FormatInt(next, 10)
if nextRound > round {
detail += " (round " + strconv.FormatInt(nextRound+1, 10) + ")"
}
if len(paged) > 0 {
detail += ": " + strings.Join(paged, ", ")
} else {
// Worth recording loudly: the rung exists, its turn came, and it woke
// nobody. That is a policy that looks configured and is not.
detail += ": nobody reachable"
}
if err := logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail); err != nil {
return err
}
return tx.Commit()
}
// escalationExhausted is the end of the line: the fallback topic, once, and a
// timeline entry saying the ladder is finished. The incident stays triggered —
// escalation running out is not the same as somebody answering.
func escalationExhausted(ctx context.Context, tx *sql.Tx, policy *escalationPolicy, incidentID int64, now time.Time) error {
detail := "escalation exhausted"
if policy.fallbackTopic != "" {
if err := enqueueNotification(ctx, tx, incidentID, nil, policy.fallbackTopic, notifyEscalated); err != nil {
return err
}
detail += ": paged " + policy.fallbackTopic
} else {
detail += ": no fallback topic configured"
}
// Level 0 again, so the sweep stops considering it. The round counter is
// left where it is, as the record of how far it got.
if _, err := tx.ExecContext(ctx,
"UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1",
incidentID); err != nil {
return err
}
return logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail)
}
// pageLevel notifies every target of one level and reports who was woken.
//
// Each target gets its own outbox row, so each gets its own Acknowledge token:
// the button in a notification must acknowledge as the person holding the
// phone, not as whoever was paged first.
func pageLevel(ctx context.Context, tx *sql.Tx, cfg NotifyConfig, policy *escalationPolicy, incidentID int64, level escalationLevel) ([]string, error) {
var paged []string
seen := map[int64]bool{}
for _, t := range level.targets {
userID := t.userID
if t.kind == "oncall" {
onCall, err := currentOnCall(ctx, tx, policy.teamID)
if err != nil {
return nil, err
}
if onCall == nil {
continue
}
userID = onCall
}
if userID == nil || seen[*userID] {
continue
}
seen[*userID] = true
var topic *string
var username string
if err := tx.QueryRowContext(ctx,
"SELECT ntfy_topic, username FROM users WHERE id = $1 AND disabled_at IS NULL",
*userID).Scan(&topic, &username); err != nil {
// A disabled or deleted account is not an error in the middle of an
// escalation: it is a target that cannot be woken, and the next
// level is the answer to that.
continue
}
if topic == nil || *topic == "" {
continue
}
if err := enqueueNotification(ctx, tx, incidentID, userID, *topic, notifyEscalated); err != nil {
return nil, err
}
paged = append(paged, username)
}
return paged, nil
}
// startEscalation puts a newly opened incident on the first rung, when its team
// has a ladder. Called from openIncident, inside the same transaction, so an
// incident is never briefly open with no escalation clock running.
func startEscalation(ctx context.Context, q querier, incidentID, teamID int64) error {
policy, err := loadEscalationPolicy(ctx, q, teamID)
if err != nil || !policy.configured() {
return err
}
_, err = q.ExecContext(ctx,
"UPDATE incidents SET escalation_level = 1, escalation_level_at = $1 WHERE id = $2",
time.Now().Unix(), incidentID)
return err
}
// stopEscalation takes an incident off the ladder. Acknowledging or resolving
// is somebody saying "I have this", and continuing to wake people after that is
// the behaviour that teaches people to ignore the tool.
func stopEscalation(ctx context.Context, q querier, incidentID int64) error {
_, err := q.ExecContext(ctx,
"UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1",
incidentID)
return err
}
// handleGetEscalation returns a team's ladder.
func handleGetEscalation(db *sql.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
teamID, ok := teamParam(w, r)
if !ok {
return
}
if !requireTeamMember(w, r, teamID) {
return
}
policy, err := loadEscalationPolicy(r.Context(), db, teamID)
if err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
respond(w, http.StatusOK, escalationResponse(policy, teamID))
}
}
type escalationLevelJSON struct {
Position int64 `json:"position"`
TimeoutSeconds int64 `json:"timeout_seconds"`
Targets []escalationTargetJSON `json:"targets"`
}
type escalationTargetJSON struct {
Kind string `json:"kind"`
UserID *int64 `json:"user_id,omitempty"`
}
type escalationJSON struct {
TeamID int64 `json:"team_id"`
RepeatCount int64 `json:"repeat_count"`
FallbackTopic string `json:"fallback_topic"`
Levels []escalationLevelJSON `json:"levels"`
}
func escalationResponse(p *escalationPolicy, teamID int64) escalationJSON {
out := escalationJSON{TeamID: teamID, Levels: []escalationLevelJSON{}}
if p == nil {
return out
}
out.RepeatCount = p.repeatCount
out.FallbackTopic = p.fallbackTopic
for _, l := range p.levels {
level := escalationLevelJSON{
Position: l.position,
TimeoutSeconds: int64(l.timeout.Seconds()),
Targets: []escalationTargetJSON{},
}
for _, t := range l.targets {
level.Targets = append(level.Targets, escalationTargetJSON{Kind: t.kind, UserID: t.userID})
}
out.Levels = append(out.Levels, level)
}
return out
}
// handleSetEscalation replaces a team's ladder wholesale.
//
// Replace rather than patch: the levels are an order, and an API that edits one
// rung has to answer what happens to the numbering of the others. Sending the
// whole ladder makes the order the client's to decide and the server's to
// store, and makes an edit atomic — there is no moment where level 2 exists
// twice.
func handleSetEscalation(db *sql.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
teamID, ok := teamParam(w, r)
if !ok {
return
}
if !requireTeamOwner(w, r, teamID) {
return
}
var req escalationJSON
if err := decodeJSON(r, &req); err != nil {
respond(w, http.StatusBadRequest, errResp("invalid request body"))
return
}
if req.RepeatCount < 0 || req.RepeatCount > 10 {
respond(w, http.StatusBadRequest, errResp("repeat_count must be between 0 and 10"))
return
}
for i, l := range req.Levels {
if l.TimeoutSeconds <= 0 {
respond(w, http.StatusBadRequest, errResp("every level needs a timeout"))
return
}
if len(l.Targets) == 0 {
// A rung that pages nobody is not a delay, it is a silence with
// a number on it.
respond(w, http.StatusBadRequest,
errResp("level "+strconv.FormatInt(int64(i+1), 10)+" has no targets"))
return
}
for _, t := range l.Targets {
switch t.Kind {
case "oncall":
if t.UserID != nil {
respond(w, http.StatusBadRequest, errResp("an oncall target takes no user_id"))
return
}
case "user":
if t.UserID == nil {
respond(w, http.StatusBadRequest, errResp("a user target needs a user_id"))
return
}
default:
respond(w, http.StatusBadRequest, errResp("target kind must be user or oncall"))
return
}
}
}
tx, err := db.BeginTx(r.Context(), nil)
if err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
defer tx.Rollback() //nolint:errcheck
if _, err := tx.ExecContext(r.Context(), `
INSERT INTO escalation_policies (team_id, repeat_count, fallback_topic, updated_at)
VALUES ($1, $2, $3, `+nowEpoch+`)
ON CONFLICT (team_id) DO UPDATE SET
repeat_count = excluded.repeat_count,
fallback_topic = excluded.fallback_topic,
updated_at = excluded.updated_at`,
teamID, req.RepeatCount, req.FallbackTopic); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
// The levels are replaced, not merged; the cascade takes the targets.
if _, err := tx.ExecContext(r.Context(),
"DELETE FROM escalation_levels WHERE team_id = $1", teamID); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
for i, l := range req.Levels {
var levelID int64
if err := tx.QueryRowContext(r.Context(), `
INSERT INTO escalation_levels (team_id, position, timeout_seconds)
VALUES ($1, $2, $3) RETURNING id`,
teamID, int64(i+1), l.TimeoutSeconds).Scan(&levelID); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
for _, t := range l.Targets {
if _, err := tx.ExecContext(r.Context(), `
INSERT INTO escalation_targets (level_id, kind, user_id)
VALUES ($1, $2, $3)`, levelID, t.Kind, t.UserID); err != nil {
// The only foreign key here is the user.
respond(w, http.StatusBadRequest, errResp("unknown user in targets"))
return
}
}
}
if err := tx.Commit(); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
policy, err := loadEscalationPolicy(r.Context(), db, teamID)
if err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
respond(w, http.StatusOK, escalationResponse(policy, teamID))
}
}
+385
View File
@@ -0,0 +1,385 @@
package api_test
import (
"net/http"
"strings"
"testing"
"time"
"git.ryuvia.com/niklas/terdut-server/internal/api"
)
// teamUser creates a user in the default team with an ntfy topic, so they can
// actually be paged.
func teamUser(t *testing.T, s *ts, username, topic string) int64 {
t.Helper()
var user struct {
ID int64 `json:"id"`
}
decode(t, s.req(t, http.MethodPost, "/api/users",
map[string]string{"username": username, "email": username + "@test.com"}), &user)
resp := s.req(t, http.MethodPost, "/api/teams/"+defaultTeam+"/members",
map[string]any{"user_id": user.ID, "role": "member"})
resp.Body.Close()
setTopic(t, s, int(user.ID), topic)
return user.ID
}
// Escalation is all timeouts, and there is no fake clock in this package. The
// tests back-date escalation_level_at instead, which is the same trick the dead
// man's switch tests use on received_at: the sweeper reads a stored timestamp,
// so moving the timestamp is moving the clock.
// ladder configures the default team with two levels: the rota first, then a
// named person, then the fallback topic.
func ladder(t *testing.T, s *ts, secondUserID int64, repeat int64, fallback string) {
t.Helper()
resp := s.req(t, http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", map[string]any{
"repeat_count": repeat,
"fallback_topic": fallback,
"levels": []map[string]any{
{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}},
{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "user", "user_id": secondUserID}}},
},
})
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("configure the ladder: %d", resp.StatusCode)
}
}
// overdue back-dates an incident's current level so its timeout has passed.
func overdue(t *testing.T, s *ts, incidentID int64) {
t.Helper()
s.exec(t, "UPDATE incidents SET escalation_level_at = $1 WHERE id = $2",
time.Now().Add(-time.Hour).Unix(), incidentID)
}
func escalationLevel(t *testing.T, s *ts, incidentID int64) (level, round int64) {
t.Helper()
if err := s.db.QueryRow(
"SELECT escalation_level, escalation_round FROM incidents WHERE id = $1",
incidentID).Scan(&level, &round); err != nil {
t.Fatalf("read escalation state: %v", err)
}
return level, round
}
// The whole point: nobody answers, so somebody else is woken.
func TestEscalation_PagesTheNextLevel(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-esc", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
// Level 1 is the rota, so the first page went to the admin.
if level, _ := escalationLevel(t, s, 1); level != 1 {
t.Fatalf("a new incident should start at level 1, got %d", level)
}
if got := f.topicsSince(t); len(got) == 0 || got[0] != "terdut-admin" {
t.Fatalf("the first page should go to the on-call user, went to %v", got)
}
// Time passes with no acknowledgement.
f.forget()
overdue(t, s, 1)
s.sweepNotify(t)
if level, _ := escalationLevel(t, s, 1); level != 2 {
t.Errorf("expected level 2, got %d", level)
}
if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-second" {
t.Errorf("level 2 should page the named user, paged %v", got)
}
// And the timeline says so, which is what somebody reads afterwards to
// understand why their phone rang at 04:00.
timeline := list(t, s.req(t, http.MethodGet, "/api/incidents/1/timeline", nil))
found := ""
for _, e := range timeline {
if e["type"] == "escalated" {
found, _ = e["detail"].(string)
}
}
if found == "" {
t.Error("the timeline should record the escalation")
} else if !strings.HasPrefix(found, "level 2") || !strings.Contains(found, "second") {
t.Errorf("the escalation entry should say which level and who: %q", found)
}
}
// Acknowledging is somebody saying "I have this". Nobody else should be woken.
func TestEscalation_AcknowledgementStopsIt(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-ack", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
s.req(t, http.MethodPost, "/api/incidents/1/acknowledge", nil).Body.Close()
if level, _ := escalationLevel(t, s, 1); level != 0 {
t.Errorf("acknowledging should take the incident off the ladder, level is %d", level)
}
f.forget()
overdue(t, s, 1) // no-op: level is 0, so there is nothing due
s.sweepNotify(t)
if got := f.topicsSince(t); len(got) != 0 {
t.Errorf("an acknowledged incident should page nobody, paged %v", got)
}
}
// Resolving stops it too, and by the same mechanism.
func TestEscalation_ResolutionStopsIt(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-res", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
s.req(t, http.MethodPost, "/api/incidents/1/resolve", nil).Body.Close()
f.forget()
overdue(t, s, 1)
s.sweepNotify(t)
if level, _ := escalationLevel(t, s, 1); level != 0 {
t.Errorf("a resolved incident should be off the ladder, level is %d", level)
}
}
// Snoozing is a deliberate "not now", so the ladder waits rather than carrying
// on without the person who asked for quiet.
func TestEscalation_SnoozePausesIt(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-snooze", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
resp := s.req(t, http.MethodPost, "/api/incidents/1/snooze", map[string]any{"duration": "1h"})
resp.Body.Close()
f.forget()
overdue(t, s, 1)
s.sweepNotify(t)
if level, _ := escalationLevel(t, s, 1); level != 1 {
t.Errorf("a snoozed incident should stay where it is, level is %d", level)
}
if got := f.topicsSince(t); len(got) != 0 {
t.Errorf("a snoozed incident should page nobody, paged %v", got)
}
// When the snooze ends, the ladder picks up where it left off.
s.exec(t, "UPDATE incidents SET snoozed_until = $1 WHERE id = 1", time.Now().Add(-time.Minute).Unix())
s.sweepNotify(t)
if level, _ := escalationLevel(t, s, 1); level != 2 {
t.Errorf("after the snooze the ladder should resume, level is %d", level)
}
}
// Running out of ladder pages the team's fallback topic once, and says so.
func TestEscalation_ExhaustionPagesTheFallback(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-end", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
overdue(t, s, 1)
s.sweepNotify(t) // level 2
f.forget()
overdue(t, s, 1)
s.sweepNotify(t) // off the end
if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-fallback" {
t.Errorf("exhaustion should page the fallback topic once, paged %v", got)
}
level, _ := escalationLevel(t, s, 1)
if level != 0 {
t.Errorf("an exhausted ladder should stop asking, level is %d", level)
}
// The incident is still open: running out of people is not an answer.
var status string
if err := s.db.QueryRow("SELECT status FROM incidents WHERE id = 1").Scan(&status); err != nil {
t.Fatal(err)
}
if status != "triggered" {
t.Errorf("exhaustion must not resolve the incident, status is %q", status)
}
}
// repeat_count walks the whole ladder again before giving up.
func TestEscalation_RepeatsTheChain(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 1, "terdut-fallback") // one extra round
postWebhook(t, s, []map[string]any{
amAlert("fp-repeat", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
overdue(t, s, 1)
s.sweepNotify(t) // level 2
f.forget()
overdue(t, s, 1)
s.sweepNotify(t) // back to level 1, round 2
level, round := escalationLevel(t, s, 1)
if level != 1 || round != 1 {
t.Errorf("expected level 1 round 1, got level %d round %d", level, round)
}
if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-admin" {
t.Errorf("the second round should start at the top again, paged %v", got)
}
}
// A team without a ladder keeps exactly the behaviour it had, and never gets
// both a reminder and an escalation for the same silence.
func TestEscalation_WithoutAPolicyRemindersStillRun(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
postWebhook(t, s, []map[string]any{
amAlert("fp-noesc", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
// Age the first notification past the repeat interval.
f.forget()
s.exec(t, "UPDATE notifications SET created_at = $1, sent_at = $1",
time.Now().Add(-time.Hour).Unix())
s.sweepNotify(t)
if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-admin" {
t.Errorf("without a ladder the reminder should still fire, paged %v", got)
}
if level, _ := escalationLevel(t, s, 1); level != 0 {
t.Errorf("an incident in a team with no ladder should not be on one, level is %d", level)
}
}
// With a ladder, reminders stop: two pages for one silence is how people learn
// to mute the tool.
func TestEscalation_WithAPolicyRemindersDoNotAlsoFire(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
second := teamUser(t, s, "second", "terdut-second")
ladder(t, s, second, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-both", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
f.forget()
// Old enough for a reminder, but not yet due for escalation.
s.exec(t, "UPDATE notifications SET created_at = $1, sent_at = $1",
time.Now().Add(-time.Hour).Unix())
s.sweepNotify(t)
if got := f.topicsSince(t); len(got) != 0 {
t.Errorf("a team with a ladder should not also get reminders, paged %v", got)
}
}
// The API refuses a ladder that cannot page anybody.
func TestEscalation_RejectsAnUnusablePolicy(t *testing.T) {
s := newTS(t)
for _, c := range []struct {
name string
body map[string]any
}{
{"a level with no targets", map[string]any{
"levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{}}},
}},
{"a level with no timeout", map[string]any{
"levels": []map[string]any{{"timeout_seconds": 0, "targets": []map[string]any{{"kind": "oncall"}}}},
}},
{"a user target with no user", map[string]any{
"levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "user"}}}},
}},
{"an unknown target kind", map[string]any{
"levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "everybody"}}}},
}},
{"an absurd repeat count", map[string]any{
"repeat_count": 99,
"levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}}},
}},
} {
resp := s.req(t, http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", c.body)
resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest {
t.Errorf("%s: expected 400, got %d", c.name, resp.StatusCode)
}
}
}
// Editing the ladder is an owner's job; reading it is any member's.
func TestEscalation_OwnerOnlyToEdit(t *testing.T) {
s := newTS(t)
_, call := member(t, s, "plain")
resp := call(http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", map[string]any{
"levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}}},
})
resp.Body.Close()
if resp.StatusCode != http.StatusForbidden {
t.Errorf("a member editing the ladder: expected 403, got %d", resp.StatusCode)
}
resp = call(http.MethodGet, "/api/teams/"+defaultTeam+"/escalation", nil)
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Errorf("a member reading the ladder: expected 200, got %d", resp.StatusCode)
}
}
// A target who cannot be woken is not a reason to stop: the next level is the
// answer to an unreachable one.
func TestEscalation_SkipsUnreachableTargets(t *testing.T) {
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute})
// Second user has no ntfy topic at all.
var user struct {
ID int64 `json:"id"`
}
decode(t, s.req(t, http.MethodPost, "/api/users",
map[string]string{"username": "silent", "email": "silent@test.com"}), &user)
s.req(t, http.MethodPost, "/api/teams/"+defaultTeam+"/members",
map[string]any{"user_id": user.ID, "role": "member"}).Body.Close()
ladder(t, s, user.ID, 0, "terdut-fallback")
postWebhook(t, s, []map[string]any{
amAlert("fp-silent", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil),
})
s.sweepNotify(t)
f.forget()
overdue(t, s, 1)
s.sweepNotify(t)
// Level 2 was entered even though it woke nobody, so the ladder keeps
// moving toward the fallback rather than stalling on a silent rung.
if level, _ := escalationLevel(t, s, 1); level != 2 {
t.Errorf("expected the ladder to advance past an unreachable target, level is %d", level)
}
if got := f.topicsSince(t); len(got) != 0 {
t.Errorf("a target with no topic should page nothing, paged %v", got)
}
}
+7
View File
@@ -235,6 +235,9 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e
if n == 0 { if n == 0 {
return false, nil return false, nil
} }
if err := stopEscalation(ctx, q, incidentID); err != nil {
return false, err
}
if err := logEvent(ctx, q, incidentID, evResolved, nil, nil, nil); err != nil { if err := logEvent(ctx, q, incidentID, evResolved, nil, nil, nil); err != nil {
return false, err return false, err
} }
@@ -260,6 +263,10 @@ func acknowledgeIncident(ctx context.Context, q querier, incidentID, userID int6
if n, _ := res.RowsAffected(); n == 0 { if n, _ := res.RowsAffected(); n == 0 {
return false, nil return false, nil
} }
// Somebody has it: stop waking anybody else.
if err := stopEscalation(ctx, q, incidentID); err != nil {
return false, err
}
return true, logEvent(ctx, q, incidentID, evAcknowledged, &userID, nil, nil) return true, logEvent(ctx, q, incidentID, evAcknowledged, &userID, nil, nil)
} }
+5
View File
@@ -243,6 +243,11 @@ func handleIncidentResolve(db *sql.DB) http.HandlerFunc {
time.Now().Unix(), incidentResolutionManual, id) { time.Now().Unix(), incidentResolutionManual, id) {
return return
} }
// A person closing an incident is the clearest possible "I have this".
if err := stopEscalation(r.Context(), db, id); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error"))
return
}
if err := logEvent(r.Context(), db, id, evResolved, &user.ID, nil, nil); err != nil { if err := logEvent(r.Context(), db, id, evResolved, &user.ID, nil, nil); err != nil {
respond(w, http.StatusInternalServerError, errResp("internal error")) respond(w, http.StatusInternalServerError, errResp("internal error"))
return return
+13 -1
View File
@@ -43,6 +43,10 @@ const (
notifyTriggered = "triggered" notifyTriggered = "triggered"
notifyReminder = "reminder" notifyReminder = "reminder"
notifyResolved = "resolved" notifyResolved = "resolved"
// notifyEscalated is a page that went out because nobody answered the last
// one. Told apart from a reminder because it goes to somebody else.
notifyEscalated = "escalated"
) )
// Timeline event types the notifier writes, so an incident's history says who // Timeline event types the notifier writes, so an incident's history says who
@@ -120,6 +124,9 @@ func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
// Exported so tests can drive a pass without waiting on the ticker. // Exported so tests can drive a pass without waiting on the ticker.
func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) { func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
enqueueReminders(ctx, db, cfg) enqueueReminders(ctx, db, cfg)
// Escalation before delivery, so a level that comes due on this tick is
// paged on this tick rather than waiting for the next one.
escalate(ctx, db, cfg)
deliverPending(ctx, db, cfg) deliverPending(ctx, db, cfg)
} }
@@ -159,7 +166,12 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
AND i.resolved_at IS NULL AND i.resolved_at IS NULL
AND i.archived_at IS NULL AND i.archived_at IS NULL
AND i.status = 'triggered' AND i.status = 'triggered'
AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2)`, AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2)
-- A team with an escalation ladder gets escalation instead. Both
-- would mean two pages for one silence, which is how people learn to
-- mute a tool.
AND NOT EXISTS (
SELECT 1 FROM escalation_levels el WHERE el.team_id = i.team_id)`,
now.Add(-repeat).Unix(), now.Unix()) now.Add(-repeat).Unix(), now.Unix())
if err != nil { if err != nil {
log.Printf("notifier: find reminders: %v", err) log.Printf("notifier: find reminders: %v", err)
+21
View File
@@ -69,6 +69,27 @@ func (f *fakeNtfy) messages() []pushed {
return append([]pushed(nil), f.got...) return append([]pushed(nil), f.got...)
} }
// topicsSince lists the topics published to since the last forget, which is how
// the escalation tests ask "who did this tick wake".
func (f *fakeNtfy) topicsSince(t *testing.T) []string {
t.Helper()
f.mu.Lock()
defer f.mu.Unlock()
out := make([]string, 0, len(f.got))
for _, m := range f.got {
out = append(out, m.Topic)
}
return out
}
// forget drops what has been published so far, so the next assertion is about
// this tick rather than the whole test.
func (f *fakeNtfy) forget() {
f.mu.Lock()
defer f.mu.Unlock()
f.got = nil
}
func (f *fakeNtfy) failWith(status int) { func (f *fakeNtfy) failWith(status int) {
f.mu.Lock() f.mu.Lock()
defer f.mu.Unlock() defer f.mu.Unlock()
+4
View File
@@ -109,6 +109,10 @@ func NewRouter(db *sql.DB, notify NotifyConfig, cfg config.Config) http.Handler
r.Post("/api/teams/{teamID}/members", handleAddTeamMember(db)) r.Post("/api/teams/{teamID}/members", handleAddTeamMember(db))
r.Delete("/api/teams/{teamID}/members/{userID}", handleRemoveTeamMember(db)) r.Delete("/api/teams/{teamID}/members/{userID}", handleRemoveTeamMember(db))
// A team's escalation ladder: who is paged when nobody answers.
r.Get("/api/teams/{teamID}/escalation", handleGetEscalation(db))
r.Put("/api/teams/{teamID}/escalation", handleSetEscalation(db))
// A team's own dead man's switches: which of its alerts are heartbeats, // A team's own dead man's switches: which of its alerts are heartbeats,
// and how long a silence has to last before somebody is paged. // and how long a silence has to last before somebody is paged.
r.Get("/api/teams/{teamID}/deadman", handleGetTeamDeadman(db)) r.Get("/api/teams/{teamID}/deadman", handleGetTeamDeadman(db))
+95
View File
@@ -0,0 +1,95 @@
-- Escalation: page somebody else when the first person does not answer.
--
-- This is the gap the whole multi-tenancy line of work was opened to close.
-- Until now an unacknowledged incident re-paged the same topic every
-- notify_repeat forever, which is a louder version of the same silence: if the
-- person on call is asleep, has no signal, or has left, nothing else happens.
--
-- Shape: one policy per team, an ordered list of levels, each level with a
-- timeout and a set of targets. When a level's timeout passes and the incident
-- is still triggered, the next level is paged. When the last level passes, the
-- chain repeats repeat_count times, and then the team's fallback topic is paged
-- once as the end of the line.
--
-- A team WITHOUT a policy keeps exactly today's behaviour: page the assignee,
-- then remind on the same topic. Escalation is opt-in per team, and the two
-- never both run for one incident -- see enqueueReminders.
CREATE TABLE escalation_policies (
-- One per team for now, hence the team as the key rather than an id with a
-- unique index: routing different alerts to different chains needs the
-- alert to carry something to route ON, which is a separate question.
team_id BIGINT PRIMARY KEY REFERENCES teams(id) ON DELETE CASCADE,
-- How many extra times to run the whole chain after it has been walked
-- once. 0 means walk it once and stop at the fallback.
repeat_count BIGINT NOT NULL DEFAULT 0 CHECK (repeat_count >= 0 AND repeat_count <= 10),
-- Where the last page goes when every level has been tried. Per team now:
-- TERDUT_NTFY_FALLBACK_TOPIC was one topic for the whole install, which in
-- a multi-team server pages the wrong people. Empty means the chain simply
-- ends.
fallback_topic TEXT NOT NULL DEFAULT '',
updated_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint
);
CREATE TABLE escalation_levels (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
team_id BIGINT NOT NULL REFERENCES escalation_policies(team_id) ON DELETE CASCADE,
-- 1-based, dense. The API rewrites the whole ladder on every edit rather
-- than patching one rung, so there is no way to leave a gap.
position BIGINT NOT NULL,
-- How long this level has to produce an acknowledgement before the next one
-- is paged. Seconds, like every other duration in this schema.
timeout_seconds BIGINT NOT NULL CHECK (timeout_seconds > 0),
UNIQUE (team_id, position)
);
-- Who a level pages. Either a named person, or whoever the team's rota says is
-- on call today -- which is the target that keeps working when the rota
-- changes and nobody remembers to edit the policy.
CREATE TABLE escalation_targets (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
level_id BIGINT NOT NULL REFERENCES escalation_levels(id) ON DELETE CASCADE,
kind TEXT NOT NULL CHECK (kind IN ('user', 'oncall')),
-- Set for kind='user', NULL for kind='oncall'.
user_id BIGINT REFERENCES users(id) ON DELETE CASCADE,
CHECK ((kind = 'user' AND user_id IS NOT NULL) OR (kind = 'oncall' AND user_id IS NULL))
);
CREATE INDEX escalation_targets_level_idx ON escalation_targets(level_id);
-- ---------------------------------------------------------------------------
-- Where an incident is in its chain.
--
-- On the incident rather than in a side table: it is read on every notifier
-- tick alongside the incident's status, and one row per incident is exactly
-- what the state is.
-- ---------------------------------------------------------------------------
-- 0 means no level has been paged yet, which is the state of every incident
-- that existed before escalation and of every incident in a team with no
-- policy. 1 is the first level.
ALTER TABLE incidents ADD COLUMN escalation_level BIGINT NOT NULL DEFAULT 0;
-- When the current level was entered, and therefore what its timeout is
-- measured from. NULL while escalation_level is 0.
ALTER TABLE incidents ADD COLUMN escalation_level_at BIGINT;
-- How many times the chain has been walked in full. Compared against the
-- policy's repeat_count.
ALTER TABLE incidents ADD COLUMN escalation_round BIGINT NOT NULL DEFAULT 0;
-- The notifier's escalation query: incidents still waiting, oldest level first.
CREATE INDEX incidents_escalation_idx
ON incidents(escalation_level_at)
WHERE resolved_at IS NULL AND status = 'triggered';
-- 'escalated' joins the outbox kinds: a page that went out because nobody
-- answered the last one, which is worth telling apart from the first page and
-- from a reminder when reading the timeline or debugging a delivery.
ALTER TABLE notifications DROP CONSTRAINT notifications_kind_check;
ALTER TABLE notifications ADD CONSTRAINT notifications_kind_check
CHECK (kind IN ('triggered', 'reminder', 'resolved', 'escalated'));