Compare commits

...

11 Commits

Author SHA1 Message Date
Niklas Ye 7efd1bbba7 Set the chart's placeholder version to 0.37.0
CI / chart (push) Successful in 1s
CI / test (push) Successful in 7s
CI / security (push) Successful in 12s
Release / test (push) Successful in 6s
Release / chart (push) Successful in 1s
Release / binaries (push) Successful in 21s
Release / image (push) Successful in 1m3s
Release / scan-image (push) Successful in 1s
2026-10-03 16:17:20 +02:00
Niklas Ye 3bf94a5d7f Default to 2 replicas and RollingUpdate now that the singleton jobs are locked
CI / chart (push) Successful in 0s
CI / test (push) Successful in 7s
CI / security (push) Successful in 12s
replicas and strategy: Recreate were the chart's only guard against the
archiver, notifier and migration races; v0.36.0 closed all three with
advisory locks and a conflict-resolving incident insert, which made
that guard redundant rather than load-bearing. Expose replicaCount
(new, no values.yaml key existed before) defaulting to 2, and switch
to strategy: RollingUpdate with no explicit maxUnavailable/maxSurge --
the 25%/25% default rounds to 0/1 at replicaCount: 2, which is already
zero-downtime.

The chart does not gate this on image.tag, so pointing it at a
pre-v0.36.0 image with the new default is a foot-gun by omission --
noted in both the values.yaml comment and the deployment.yaml comment,
not guarded in code, same as the chart does for every other
version-coupled assumption today.

Co-authored-by: Claude <noreply@anthropic.com>
2026-10-03 12:30:27 +02:00
Niklas Ye 5c4e0bdd0e Set the chart's placeholder version to 0.36.0
CI / chart (push) Successful in 1s
CI / security (push) Successful in 16s
CI / test (push) Successful in 4m50s
Release / test (push) Successful in 7s
Release / chart (push) Successful in 2s
Release / binaries (push) Successful in 48s
Release / image (push) Successful in 1m12s
Release / scan-image (push) Successful in 2s
2026-10-03 12:03:28 +02:00
Niklas Ye 9e5b085d8b Resolve the new-incident insert conflict instead of dropping the payload
CI / chart (push) Successful in 1s
CI / security (push) Successful in 20s
CI / test (push) Has been cancelled
openIncident's INSERT had no ON CONFLICT clause, relying entirely on
incidentForGroup's earlier SELECT to avoid a duplicate. On more than
one replica, two webhook deliveries for the very first occurrence of
a brand-new groupKey can both pass that SELECT before either INSERTs;
the loser then hit incidents_open_group_key_idx's unique violation,
which rolled back its whole transaction — including that payload's
alert upserts, done earlier in the same transaction. ingest's error is
only logged and receiveWebhook answers 200 regardless, so nothing
retried it: the loser's alerts silently never existed.

Add ON CONFLICT (team_id, group_key) WHERE resolved_at IS NULL DO
NOTHING to the INSERT, matching the partial unique index. Postgres
only resolves that conflict after the winning transaction commits (or
rolls back), so by the time RETURNING comes back empty,
existingOpenIncident's follow-up SELECT is guaranteed to see the
winner's row. The loser attaches to it instead of failing outright,
and the rest of its payload commits normally. Covers both callers,
since the dead man's switch sweeper shares this same function.

New test (package api_test, fires N webhook deliveries for one
groupKey from a synchronized start with distinct fingerprints, so
they aren't accidentally serialized by upsertAlerts' own per-
fingerprint lock) confirmed meaningful: with the ON CONFLICT clause
reverted, it fails 10/10 on a missing alert fingerprint; restored, 0/10.
Note while building it: "exactly one incident" alone cannot
distinguish fixed from broken, since the DB's own unique index already
guarantees that either way — the real signal is the loser's payload
surviving.

Chart comment updated: all three of the chart's original reasons for
Recreate are now addressed in code, though replicas stays at 1 and the
strategy stays Recreate pending a deliberate decision to raise it.

Co-authored-by: Claude <noreply@anthropic.com>
2026-10-03 11:58:39 +02:00
Niklas Ye 0050738ca0 Advisory-lock DB migrations against concurrent replica startup
Migrate's check-then-apply loop against schema_migrations had no
locking: two replicas booting at once against a fresh or
partially-migrated database could both pass the "not yet applied"
check for the same file and race applying it, crashing whichever lost
the duplicate-key insert (confirmed: reverting the lock fails the new
test 10/10 on a duplicate-key violation, racing as early as the
CREATE TABLE IF NOT EXISTS schema_migrations statement itself).

Hold a Postgres advisory lock for Migrate's whole run, on a dedicated
connection reserved via db.Conn so lock and unlock happen on the same
session. Blocking (pg_advisory_lock), unlike the archiver/notifier's
pg_try_advisory_lock: on boot there's no later tick to defer to, so a
second replica should wait for the first to finish migrating rather
than skip ahead.

Adds internal/db's first test file, exercising two concurrent Migrate
calls against a fresh schema.

Still open: the new-incident-insert race on a webhook for a brand-new
groupKey, noted in the chart's updated comment. Login rate limiting
staying in-process, diluted across replicas, is an accepted tradeoff.

Co-authored-by: Claude <noreply@anthropic.com>
2026-10-03 11:43:09 +02:00
Niklas Ye 42180948d1 Guard the archiver and notifier passes with a Postgres advisory lock
Both background loops run unconditionally on every instance with no
coordination between them, which the chart's replicas: 1 + strategy:
Recreate exists specifically to paper over: with more than one replica,
every one of them would sweep and deliver notifications independently,
and two overlapping during a rollout would both page for the same
incident.

Add withAdvisoryLock, which takes a Postgres advisory lock on a
dedicated connection and runs a pass only if it gets the lock,
otherwise skipping until the next tick. Wire StartArchiver and
StartNotifier through it with their own lock keys, so Sweep and
NotifySweep themselves are untouched and every existing test calling
them directly keeps working unchanged.

This also closes the notifier's double-delivery race in passing: two
replicas can no longer both be inside deliverPending at once, since
only one can hold notifierLockKey at a time.

Deliberately not addressed here, and still blocking a replica count
above 1: the in-memory login rate limiter, the unlocked migration
runner, and the new-incident-insert race on a webhook for a brand-new
groupKey. Noted in the updated chart comment.

Co-authored-by: Claude <noreply@anthropic.com>
2026-10-03 11:36:48 +02:00
Niklas Ye c83c7c2a8b Set the chart's placeholder version to 0.35.1
CI / chart (push) Successful in 2s
CI / security (push) Successful in 20s
CI / test (push) Successful in 4m45s
Release / test (push) Successful in 3s
Release / chart (push) Successful in 2s
Release / binaries (push) Successful in 50s
Release / image (push) Successful in 1m21s
Release / scan-image (push) Successful in 3s
Cosmetic, as before (497086c, 4358e84): make helm-package passes
--version and --app-version from the tag, so this decides nothing
about what gets published. Kept in step so the tree heading for
v0.35.1 doesn't say 0.35.0 to a reader who hasn't yet seen the tag.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-03 09:39:01 +02:00
niklas 43beda9a30 Merge pull request 'Fix the queue chip fade falling short of the right edge' (#30) from fix-chips-fade-edge into main
CI / chart (push) Successful in 1s
CI / security (push) Successful in 21s
CI / test (push) Has been cancelled
2026-10-03 07:36:12 +00:00
Niklas Ye 710521a73c Fix the queue chip fade falling short of the right edge
CI / chart (pull_request) Successful in 1s
CI / security (pull_request) Successful in 16s
CI / test (pull_request) Successful in 5m13s
position: sticky, as a flex item of the row it's pinning itself
against, interacted with that row's gap and its own negative margin in
a way that landed it short of the true edge -- visibly, a sliver of
the next chip stayed poking out past where the fade should have
covered it, which is the "ends before the screen edge" bug reported
against the release.

Replaced with the simpler, better-supported pattern for this: an
absolutely positioned overlay against a position:relative, overflow:
auto parent. Unlike a sticky descendant, an absolutely positioned one
is resolved against the parent's own (non-scrolling) box, so it stays
flush with the real edge regardless of scroll position, without the
flex-gap/margin interaction that caused this.

Verified visually: a standalone reproduction of both versions,
screenshotted with Chromium's headless_shell (no browser automation
tool available in this environment, but the binary's right there) --
the old version shows the next chip's edge peeking past the fade, the
new one doesn't.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-03 09:33:33 +02:00
niklas d9492913ed Merge pull request 'Fix Stats/Admin/Account showing in the bottom bar too' (#29) from fix-nav-secondary-cascade into main
CI / chart (push) Successful in 0s
CI / test (push) Successful in 8s
CI / security (push) Successful in 14s
Reviewed-on: #29
2026-10-03 07:30:04 +00:00
Niklas Ye 1770e5d945 Fix Stats/Admin/Account showing in the bottom bar too
CI / chart (pull_request) Successful in 1s
CI / security (pull_request) Successful in 16s
CI / test (pull_request) Successful in 5m2s
.nav-link-secondary's display:none sat before .nav-link's own
display:flex in the file. Both are single-class selectors, so they tie
on specificity, and a tie is broken by which one comes later in the
file -- not by which class the element happens to carry. .nav-link's
declaration, being later, won for every element wearing both classes,
so Stats/Admin/Account rendered as three extra tabs on the phone bar
instead of folding into "More" as intended.

Moved the rule below .nav-link instead of changing either declaration,
since nothing about the values was wrong -- only their order was.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-03 09:22:48 +02:00
12 changed files with 527 additions and 17 deletions
+2 -2
View File
@@ -15,5 +15,5 @@ type: application
# appVersion and image.tag in values.yaml no longer agree, and that is not an oversight: # appVersion and image.tag in values.yaml no longer agree, and that is not an oversight:
# image.tag stays "latest", which is what a local install actually pulls. appVersion is # image.tag stays "latest", which is what a local install actually pulls. appVersion is
# metadata and drives nothing. # metadata and drives nothing.
version: 0.35.0 version: 0.37.0
appVersion: "v0.35.0" appVersion: "v0.37.0"
@@ -6,15 +6,19 @@ metadata:
labels: labels:
{{- include "terdut-server.labels" . | nindent 4 }} {{- include "terdut-server.labels" . | nindent 4 }}
spec: spec:
replicas: 1 replicas: {{ .Values.replicaCount }}
selector: selector:
matchLabels: matchLabels:
{{- include "terdut-server.selectorLabels" . | nindent 6 }} {{- include "terdut-server.selectorLabels" . | nindent 6 }}
# Recreate, not RollingUpdate, even though the PVC that forced it is gone: the # RollingUpdate, not Recreate: the sweeper, notifier and migration runner
# sweeper and the notifier are unsynchronised singletons, and two replicas # each take a Postgres advisory lock around their own pass, and new-incident
# overlapping during a rollout would both page for the same incident. # creation on the first webhook for a brand-new groupKey resolves its own
# insert conflict -- so two replicas overlapping during a rollout no longer
# double-page, race a migration, or drop a webhook payload (v0.36.0). No
# explicit maxUnavailable/maxSurge: the 25%/25% default rounds to 0/1 at
# replicaCount: 2, which is zero-downtime already.
strategy: strategy:
type: Recreate type: RollingUpdate
template: template:
metadata: metadata:
labels: labels:
+7
View File
@@ -1,3 +1,10 @@
# Safe above 1 since v0.36.0: the sweeper, notifier and migration runner each
# take a Postgres advisory lock around their own pass, and a webhook that
# loses the race to open a brand-new incident attaches to the winner's row
# instead of dropping its payload. An image older than v0.36.0 does not have
# these guards -- do not raise this against one.
replicaCount: 2
networking: networking:
hostname: "terdut.example.com" hostname: "terdut.example.com"
servicePort: 8080 servicePort: 8080
+122
View File
@@ -0,0 +1,122 @@
package api
// This file is internal (package api, not api_test) because withAdvisoryLock is
// unexported and these tests exercise its locking semantics directly rather than
// through the full StartArchiver/StartNotifier loop, which would make the "does
// not run while held" case timing-dependent instead of deterministic. It opens a
// plain connection to TERDUT_TEST_DSN rather than reusing testdb_test.go's
// newTestDB, since that helper lives in the separate, already-compiled
// api_test package and a Postgres advisory lock needs no schema or migration
// to exercise.
import (
"context"
"database/sql"
"os"
"testing"
_ "github.com/jackc/pgx/v5/stdlib"
)
// advisoryTestDB opens a plain, unmigrated connection to the test database. An
// unset DSN fails rather than skips, matching testdb_test.go's rationale: a
// suite that quietly tests nothing is worse than one that does not run.
func advisoryTestDB(t *testing.T) *sql.DB {
t.Helper()
dsn := os.Getenv("TERDUT_TEST_DSN")
if dsn == "" {
t.Fatalf("TERDUT_TEST_DSN is not set: these tests need Postgres.\n" +
"Run `make test-db` for a local one, then\n" +
" export TERDUT_TEST_DSN=postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable")
}
db, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatalf("connect to TERDUT_TEST_DSN: %v", err)
}
t.Cleanup(func() { db.Close() })
return db
}
func TestWithAdvisoryLock_RunsWhenFree(t *testing.T) {
db := advisoryTestDB(t)
ctx := context.Background()
ran := false
withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })
if !ran {
t.Fatal("fn did not run although the lock was free")
}
}
func TestWithAdvisoryLock_SkipsWhileHeldElsewhere(t *testing.T) {
db := advisoryTestDB(t)
ctx := context.Background()
// Hold the lock on a connection of our own, standing in for another
// replica mid-pass.
holder, err := db.Conn(ctx)
if err != nil {
t.Fatalf("acquire holder connection: %v", err)
}
defer holder.Close()
if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_lock($1)", archiverLockKey); err != nil {
t.Fatalf("pre-acquire lock: %v", err)
}
ran := false
withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })
if ran {
t.Fatal("fn ran although another connection already held the lock")
}
if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", archiverLockKey); err != nil {
t.Fatalf("release held lock: %v", err)
}
// Now that the holder released it, the next caller should get it.
ran = false
withAdvisoryLock(ctx, db, archiverLockKey, "test", func() { ran = true })
if !ran {
t.Fatal("fn did not run after the other connection released the lock")
}
}
func TestWithAdvisoryLock_ReleasesAfterFnReturns(t *testing.T) {
db := advisoryTestDB(t)
ctx := context.Background()
withAdvisoryLock(ctx, db, notifierLockKey, "test", func() {})
// If the first call had leaked the lock, this one would see it held and
// skip, leaving ran false.
ran := false
withAdvisoryLock(ctx, db, notifierLockKey, "test", func() { ran = true })
if !ran {
t.Fatal("fn did not run on a later call: the earlier call leaked its lock")
}
}
func TestWithAdvisoryLock_KeysAreIndependent(t *testing.T) {
db := advisoryTestDB(t)
ctx := context.Background()
holder, err := db.Conn(ctx)
if err != nil {
t.Fatalf("acquire holder connection: %v", err)
}
defer holder.Close()
if _, err := holder.ExecContext(ctx, "SELECT pg_advisory_lock($1)", archiverLockKey); err != nil {
t.Fatalf("pre-acquire archiver lock: %v", err)
}
defer holder.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", archiverLockKey)
// Holding archiverLockKey must not block notifierLockKey.
ran := false
withAdvisoryLock(ctx, db, notifierLockKey, "test", func() { ran = true })
if !ran {
t.Fatal("fn did not run under a different key although only archiverLockKey was held")
}
}
+35 -1
View File
@@ -376,6 +376,17 @@ func incidentForGroup(ctx context.Context, tx *sql.Tx, notify NotifyConfig, team
// its own. Hence the querier rather than a *sql.Tx. A nil severity leaves the // its own. Hence the querier rather than a *sql.Tx. A nil severity leaves the
// column for refreshSeverity to fill from the member alerts; the sweeper passes // column for refreshSeverity to fill from the member alerts; the sweeper passes
// one because its incidents have no members to derive it from. // one because its incidents have no members to derive it from.
//
// Both callers get here only after their own SELECT found no open incident for
// this group_key — but on more than one replica, two webhook deliveries for the
// very first occurrence of a brand-new group_key can both pass that SELECT
// before either INSERTs. ON CONFLICT DO NOTHING against
// incidents_open_group_key_idx is what makes the loser's INSERT a no-op instead
// of a unique-violation error that would otherwise roll back its entire
// payload; existingOpenIncident then hands it the winner's row. Postgres
// resolves that conflict only once the winner's transaction has committed (or
// rolled back), so by the time this RETURNING comes back empty, the winner's
// row is guaranteed visible to that follow-up SELECT.
func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID int64, groupKey, title string, groupLabels map[string]string, severity *string) (int64, error) { func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID int64, groupKey, title string, groupLabels map[string]string, severity *string) (int64, error) {
onCall, err := currentOnCall(ctx, q, teamID) onCall, err := currentOnCall(ctx, q, teamID)
if err != nil { if err != nil {
@@ -391,10 +402,18 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID in
err = q.QueryRowContext(ctx, ` err = q.QueryRowContext(ctx, `
INSERT INTO incidents (team_id, group_key, title, group_labels, signature, status, severity, triggered_at, assigned_to) INSERT INTO incidents (team_id, group_key, title, group_labels, signature, status, severity, triggered_at, assigned_to)
VALUES ($1, $2, $3, $4::jsonb, $5, 'triggered', $6, $7, $8) VALUES ($1, $2, $3, $4::jsonb, $5, 'triggered', $6, $7, $8)
ON CONFLICT (team_id, group_key) WHERE resolved_at IS NULL DO NOTHING
RETURNING id`, RETURNING id`,
teamID, groupKey, title, string(labelsJSON), incidentSignature(groupLabels, title), severity, teamID, groupKey, title, string(labelsJSON), incidentSignature(groupLabels, title), severity,
time.Now().Unix(), onCall).Scan(&id) time.Now().Unix(), onCall).Scan(&id)
if err != nil { switch {
case err == sql.ErrNoRows:
// Lost the race: someone else's incident for this group_key exists now.
// Everything below — the trigger event, assignment, page, escalation
// clock — already happened for that row when it was created; attach to
// it rather than fail this call (and the whole payload) outright.
return existingOpenIncident(ctx, q, teamID, groupKey)
case err != nil:
return 0, err return 0, err
} }
@@ -423,6 +442,21 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID in
return id, nil return id, nil
} }
// existingOpenIncident looks up the open incident openIncident's own INSERT just
// lost a conflict against — the same lookup incidentForGroup does before ever
// calling openIncident, repeated here for the caller that arrived second.
func existingOpenIncident(ctx context.Context, q querier, teamID int64, groupKey string) (int64, error) {
var id int64
err := q.QueryRowContext(ctx,
"SELECT id FROM incidents WHERE team_id = $1 AND group_key = $2 AND resolved_at IS NULL",
teamID, groupKey,
).Scan(&id)
if err != nil {
return 0, err
}
return id, nil
}
// linkAlert adds an alert to an incident, emitting a timeline entry only the // linkAlert adds an alert to an incident, emitting a timeline entry only the
// first time. Re-sends of an already-linked alert are silent. // first time. Re-sends of an already-linked alert are silent.
func linkAlert(ctx context.Context, tx *sql.Tx, incidentID, alertID int64) error { func linkAlert(ctx context.Context, tx *sql.Tx, incidentID, alertID int64) error {
+112
View File
@@ -0,0 +1,112 @@
package api_test
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"sync"
"testing"
)
// TestWebhook_ConcurrentFirstOccurrenceOpensOneIncident reproduces two
// replicas racing the very first webhook delivery for a brand-new group_key:
// both see no open incident yet (incidentForGroup's own SELECT finds
// nothing) and race openIncident's INSERT.
//
// The DB's own unique index already guarantees at most one incident either
// way, with or without this fix — so "exactly one incident" alone cannot
// tell a fixed run from a broken one. What ON CONFLICT handling actually
// changes is what happens to the *loser*: before it, the loser's INSERT hit
// incidents_open_group_key_idx's unique violation, which — since
// upsertAlerts ran earlier in that same transaction — rolled back its whole
// payload, alert insert included. ingest's error is only logged and
// receiveWebhook answers 200 regardless, so nothing ever retried it: the
// loser's alert silently never existed. That is the regression signal this
// test checks — every caller's fingerprint must show up in /api/alerts, not
// just the winner's.
func TestWebhook_ConcurrentFirstOccurrenceOpensOneIncident(t *testing.T) {
s := newTS(t)
const callers = 8
const groupKey = "race-group"
// Every caller needs its own fingerprint. A shared one would serialize all
// of them at upsertAlerts' own ON CONFLICT (team_id, fingerprint) row lock,
// long before any of them reached incidentForGroup — which would hide the
// very race this test exists to force.
bodies := make([][]byte, callers)
for i := range callers {
payload := map[string]any{
"version": "4",
"status": "firing",
"groupKey": groupKey,
"groupLabels": map[string]string{"alertname": "RaceAlert"},
"alerts": []map[string]any{amAlert(fmt.Sprintf("fp-race-%d", i), "RaceAlert", "firing",
"2026-05-20T10:00:00Z", "0001-01-01T00:00:00Z", nil)},
}
bodies[i], _ = json.Marshal(payload)
}
// A start line, so every request is fired as close to simultaneously as
// goroutine scheduling allows, rather than trickling out one dial at a
// time — the race window is the gap between incidentForGroup's SELECT and
// openIncident's INSERT, which a staggered start could easily miss.
var ready sync.WaitGroup
start := make(chan struct{})
statuses := make([]int, callers)
var wg sync.WaitGroup
for i := range callers {
ready.Add(1)
wg.Add(1)
go func(i int) {
defer wg.Done()
ready.Done()
<-start
resp, err := http.Post(s.URL+"/api/integrations/"+s.ingestKey+"/alertmanager",
"application/json", bytes.NewReader(bodies[i]))
if err != nil {
t.Errorf("post webhook #%d: %v", i, err)
return
}
defer resp.Body.Close()
statuses[i] = resp.StatusCode
}(i)
}
ready.Wait()
close(start)
wg.Wait()
for i, code := range statuses {
if code != http.StatusOK {
t.Errorf("webhook #%d returned %d, want 200", i, code)
}
}
var matched []any
for _, inc := range listIncidents(t, s, "") {
if inc["group_key"] == groupKey {
matched = append(matched, inc["id"])
}
}
if len(matched) != 1 {
t.Fatalf("expected exactly 1 incident for group_key %q after %d concurrent deliveries, got %d: %v",
groupKey, callers, len(matched), matched)
}
var alerts []map[string]any
decode(t, s.req(t, http.MethodGet, "/api/alerts", nil), &alerts)
seen := map[string]bool{}
for _, a := range alerts {
if fp, ok := a["fingerprint"].(string); ok {
seen[fp] = true
}
}
for i := range callers {
fp := fmt.Sprintf("fp-race-%d", i)
if !seen[fp] {
t.Errorf("alert %q is missing: its delivery's whole payload was silently rolled back "+
"when it lost the race for the incident", fp)
}
}
}
+18 -2
View File
@@ -15,6 +15,12 @@ const (
// expiryGrace absorbs clock skew and notification latency before an alert // expiryGrace absorbs clock skew and notification latency before an alert
// whose ends_at watermark has passed is treated as stale. // whose ends_at watermark has passed is treated as stale.
expiryGrace = 5 * time.Minute expiryGrace = 5 * time.Minute
// archiverLockKey is the Postgres advisory lock the sweeper takes for the
// duration of each pass, so that running more than one replica does not run
// the sweep concurrently on all of them. Its value has no meaning beyond
// being distinct from notifierLockKey.
archiverLockKey int64 = 7265_0001
) )
// StartArchiver runs the alert sweeper until ctx is cancelled, starting with an // StartArchiver runs the alert sweeper until ctx is cancelled, starting with an
@@ -23,15 +29,25 @@ const (
// the fallback, not the setting: each pass reads the current value from the // the fallback, not the setting: each pass reads the current value from the
// settings table, so an administrator's change takes effect on the next tick // settings table, so an administrator's change takes effect on the next tick
// instead of at the next restart. // instead of at the next restart.
//
// Each pass runs under archiverLockKey (see withAdvisoryLock), so that on more
// than one replica only whichever instance's tick takes the lock first actually
// sweeps; the rest skip that tick rather than racing the same pass.
func StartArchiver(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Duration, notify NotifyConfig) { func StartArchiver(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Duration, notify NotifyConfig) {
ticker := time.NewTicker(sweepInterval) ticker := time.NewTicker(sweepInterval)
defer ticker.Stop() defer ticker.Stop()
Sweep(ctx, db, archiveAfter, staleAfter, notify) sweep := func() {
withAdvisoryLock(ctx, db, archiverLockKey, "sweeper", func() {
Sweep(ctx, db, archiveAfter, staleAfter, notify)
})
}
sweep()
for { for {
select { select {
case <-ticker.C: case <-ticker.C:
Sweep(ctx, db, archiveAfter, staleAfter, notify) sweep()
case <-ctx.Done(): case <-ctx.Done():
return return
} }
+39
View File
@@ -1,8 +1,11 @@
package api package api
import ( import (
"context"
"database/sql"
"encoding/json" "encoding/json"
"errors" "errors"
"log"
"net/http" "net/http"
"strconv" "strconv"
"strings" "strings"
@@ -77,3 +80,39 @@ func decodeJSON(r *http.Request, v any) error {
func errResp(msg string) map[string]string { func errResp(msg string) map[string]string {
return map[string]string{"error": msg} return map[string]string{"error": msg}
} }
// withAdvisoryLock runs fn only if it can take the named Postgres advisory lock on a
// dedicated connection, and skips fn otherwise. This is what keeps the archiver and
// notifier safe to run on more than one replica: whichever instance's tick gets there
// first does the work; the rest see the lock held and simply wait for their next tick
// instead of running the same pass concurrently.
//
// pg_try_advisory_lock is session-scoped, so taking and releasing it must happen on the
// same connection, reserved via db.Conn rather than borrowed from the pool's shared
// connections fn itself may use — and released (unlocked, then closed) before returning,
// since a session lock otherwise outlives this call and leaks onto whatever reuses the
// pooled connection next.
func withAdvisoryLock(ctx context.Context, db *sql.DB, key int64, name string, fn func()) {
conn, err := db.Conn(ctx)
if err != nil {
log.Printf("%s: advisory lock: acquire connection: %v", name, err)
return
}
defer conn.Close()
var locked bool
if err := conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", key).Scan(&locked); err != nil {
log.Printf("%s: advisory lock: %v", name, err)
return
}
if !locked {
return // another replica is already running this pass
}
defer func() {
if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", key); err != nil {
log.Printf("%s: advisory unlock: %v", name, err)
}
}()
fn()
}
+19 -2
View File
@@ -38,6 +38,13 @@ const (
ackTokenTTL = 24 * time.Hour ackTokenTTL = 24 * time.Hour
) )
// notifierLockKey is the Postgres advisory lock the notifier takes for the
// duration of each pass, so that running more than one replica does not
// deliver (or double-deliver) the same notification from more than one of
// them at once. Its value has no meaning beyond being distinct from
// archiverLockKey.
const notifierLockKey int64 = 7265_0002
// Notification kinds, recording why a push was sent. // Notification kinds, recording why a push was sent.
const ( const (
notifyTriggered = "triggered" notifyTriggered = "triggered"
@@ -97,6 +104,10 @@ var notifyClient = &http.Client{Timeout: 10 * time.Second}
// StartNotifier delivers queued notifications until ctx is cancelled, starting // StartNotifier delivers queued notifications until ctx is cancelled, starting
// with an immediate pass so a restart flushes whatever the last one left behind. // with an immediate pass so a restart flushes whatever the last one left behind.
//
// Each pass runs under notifierLockKey (see withAdvisoryLock), so that on more
// than one replica only whichever instance's tick takes the lock first actually
// delivers; the rest skip that tick rather than racing the same pass.
func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) { func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
if !cfg.enabled() { if !cfg.enabled() {
log.Print("notifier: disabled (no ntfy URL configured)") log.Print("notifier: disabled (no ntfy URL configured)")
@@ -107,11 +118,17 @@ func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
ticker := time.NewTicker(notifyInterval) ticker := time.NewTicker(notifyInterval)
defer ticker.Stop() defer ticker.Stop()
NotifySweep(ctx, db, cfg) sweep := func() {
withAdvisoryLock(ctx, db, notifierLockKey, "notifier", func() {
NotifySweep(ctx, db, cfg)
})
}
sweep()
for { for {
select { select {
case <-ticker.C: case <-ticker.C:
NotifySweep(ctx, db, cfg) sweep()
case <-ctx.Done(): case <-ctx.Done():
return return
} }
+27
View File
@@ -1,6 +1,7 @@
package db package db
import ( import (
"context"
"database/sql" "database/sql"
"embed" "embed"
"fmt" "fmt"
@@ -62,6 +63,16 @@ func Open(dsn string) (*sql.DB, error) {
} }
} }
// migrationLockKey is the Postgres advisory lock Migrate holds for its whole
// run. Two replicas starting at once would otherwise race the check-then-apply
// loop below against schema_migrations: the loser could crash on a
// duplicate-key insert, or contend with the winner's uncommitted DDL. Blocking
// (pg_advisory_lock, not pg_try_advisory_lock as the archiver and notifier
// use): on boot there is no later tick to defer to, so the right behaviour is
// to wait for the other replica to finish migrating, not to skip ahead and
// start serving against an unmigrated schema.
const migrationLockKey int64 = 7265_0003
// Migrate applies every embedded migration that has not been applied yet, in // Migrate applies every embedded migration that has not been applied yet, in
// filename order, recording each in schema_migrations. // filename order, recording each in schema_migrations.
// //
@@ -69,6 +80,22 @@ func Open(dsn string) (*sql.DB, error) {
// migration that failed half way used to leave the schema in whatever state it // migration that failed half way used to leave the schema in whatever state it
// had reached. Postgres has transactional DDL, so the rollback is real. // had reached. Postgres has transactional DDL, so the rollback is real.
func Migrate(db *sql.DB) error { func Migrate(db *sql.DB) error {
ctx := context.Background()
conn, err := db.Conn(ctx)
if err != nil {
return fmt.Errorf("migrate: acquire connection: %w", err)
}
defer conn.Close()
if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil {
return fmt.Errorf("migrate: acquire advisory lock: %w", err)
}
defer func() {
if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", migrationLockKey); err != nil {
log.Printf("migrate: release advisory lock: %v", err)
}
}()
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations ( if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations (
version TEXT PRIMARY KEY, version TEXT PRIMARY KEY,
applied_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint applied_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint
+125
View File
@@ -0,0 +1,125 @@
package db_test
import (
"database/sql"
"fmt"
"net/url"
"os"
"strings"
"sync"
"testing"
"git.ryuvia.com/niklas/terdut-server/internal/db"
_ "github.com/jackc/pgx/v5/stdlib"
)
// TERDUT_TEST_DSN must point at a database the test role may create schemas
// in; see internal/api/testdb_test.go for the fuller rationale this mirrors.
// An unset DSN fails rather than skips, deliberately.
const testDSNEnv = "TERDUT_TEST_DSN"
// TestMigrate_ConcurrentCallersDoNotRace reproduces two replicas starting at
// once against a brand-new, unmigrated schema: both call db.Migrate at the
// same time. Before migrationLockKey, the loser could crash on a
// duplicate-key insert into schema_migrations, or contend with the winner's
// uncommitted DDL; with the advisory lock, one blocks until the other
// finishes and both return cleanly.
func TestMigrate_ConcurrentCallersDoNotRace(t *testing.T) {
dsn := os.Getenv(testDSNEnv)
if dsn == "" {
t.Fatalf("%s is not set: these tests need Postgres.\n"+
"Run `make test-db` for a local one, then\n"+
" export %s=postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable",
testDSNEnv, testDSNEnv)
}
schema := fmt.Sprintf("migrate_race_%d", os.Getpid())
admin, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatalf("connect to %s: %v", testDSNEnv, err)
}
defer admin.Close()
if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
t.Fatalf("create schema %s: %v", schema, err)
}
t.Cleanup(func() {
cleanup, err := sql.Open("pgx", dsn)
if err != nil {
return
}
defer cleanup.Close()
if _, err := cleanup.Exec("DROP SCHEMA " + schema + " CASCADE"); err != nil {
t.Logf("drop schema %s: %v", schema, err)
}
})
scoped := withSearchPath(dsn, schema)
const callers = 2
errs := make([]error, callers)
var wg sync.WaitGroup
for i := range callers {
wg.Add(1)
go func(i int) {
defer wg.Done()
database, err := db.Open(scoped)
if err != nil {
errs[i] = fmt.Errorf("open: %w", err)
return
}
defer database.Close()
errs[i] = db.Migrate(database)
}(i)
}
wg.Wait()
for i, err := range errs {
if err != nil {
t.Fatalf("Migrate #%d: %v", i, err)
}
}
entries, err := os.ReadDir("migrations")
if err != nil {
t.Fatalf("read migrations dir: %v", err)
}
var want int
for _, e := range entries {
if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
want++
}
}
check, err := sql.Open("pgx", scoped)
if err != nil {
t.Fatalf("connect for verification: %v", err)
}
defer check.Close()
var got int
if err := check.QueryRow("SELECT COUNT(*) FROM schema_migrations").Scan(&got); err != nil {
t.Fatalf("count schema_migrations: %v", err)
}
if got != want {
t.Fatalf("schema_migrations has %d row(s) after two concurrent Migrate calls, want %d (one per migration file, no duplicates)", got, want)
}
}
// withSearchPath pins a DSN to one schema. Copied from
// internal/api/testdb_test.go rather than shared: that helper lives in the
// api_test package, a separate compiled package this one cannot import.
func withSearchPath(dsn, schema string) string {
opt := "-csearch_path=" + schema
if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") {
u, err := url.Parse(dsn)
if err == nil {
q := u.Query()
q.Set("options", opt)
u.RawQuery = q.Encode()
return u.String()
}
}
return dsn + " options='" + opt + "'"
}
+12 -5
View File
@@ -255,7 +255,6 @@ input:focus, textarea:focus { outline: none; border-color: var(--accent); box-sh
/* Hidden here (shown from 900px below): on the phone bar the team switcher /* Hidden here (shown from 900px below): on the phone bar the team switcher
lives in the topbar instead, as #team-selector-mobile. */ lives in the topbar instead, as #team-selector-mobile. */
.nav-team-selector { display: none; } .nav-team-selector { display: none; }
.nav-link-secondary { display: none; }
.nav-link { .nav-link {
position: relative; position: relative;
/* flex: 1 spreads the tabs evenly across the bar's width; the desktop /* flex: 1 spreads the tabs evenly across the bar's width; the desktop
@@ -267,6 +266,10 @@ input:focus, textarea:focus { outline: none; border-color: var(--accent); box-sh
what stops six tabs widening the bar past the screen. */ what stops six tabs widening the bar past the screen. */
min-width: 0; padding: 0 2px; min-width: 0; padding: 0 2px;
} }
/* Must come after .nav-link above: same specificity (one class each), so
whichever is later in the file wins for an element wearing both classes,
and this needs to beat .nav-link's display:flex here on the phone bar. */
.nav-link-secondary { display: none; }
.nav-label { .nav-label {
max-width: 100%; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; max-width: 100%; overflow: hidden; text-overflow: ellipsis; white-space: nowrap;
} }
@@ -323,6 +326,7 @@ input:focus, textarea:focus { outline: none; border-color: var(--accent); box-sh
/* ---------- chips ---------- */ /* ---------- chips ---------- */
.chips { .chips {
position: relative;
display: flex; gap: 6px; display: flex; gap: 6px;
padding: 12px 16px 8px; padding: 12px 16px 8px;
overflow-x: auto; scrollbar-width: none; overflow-x: auto; scrollbar-width: none;
@@ -338,11 +342,14 @@ input:focus, textarea:focus { outline: none; border-color: var(--accent); box-sh
} }
.chip[aria-selected="true"] { background: var(--text); border-color: var(--text); color: var(--bg); } .chip[aria-selected="true"] { background: var(--text); border-color: var(--text); color: var(--bg); }
.chip .count { margin-left: 4px; opacity: 0.7; } .chip .count { margin-left: 4px; opacity: 0.7; }
/* Pinned to the visible right edge of the scrolling row (sticky, not /* An overlay, not a flex item: absolute against .chips' own (non-scrolling)
absolute, so it tracks the scroll position rather than the content). */ box stays flush with its real right edge regardless of scroll position,
which turned out not to be true of position:sticky here — as a flex
item, its sticky offset interacted with the row's gap and its own
negative margin, landing short of the edge by about one gap's width. */
.chips-fade { .chips-fade {
position: sticky; right: -1px; flex: none; position: absolute; top: 0; right: 0; bottom: 0;
width: 24px; margin-left: -24px; width: 24px;
background: linear-gradient(to right, transparent, var(--bg)); background: linear-gradient(to right, transparent, var(--bg));
pointer-events: none; pointer-events: none;
} }