amadan

carlos / platform Public

Merge branch 'fix/timer-rearm-completion' into main

paul@keymail.dev 203f67a9e2cd057fc9ccbab83ef9eddcdee6afd5 parent 1643418parent 5c6edc6
40 files changed, +1621 −74
  • apimode.go +19 −1
  • docs/site/cli.md +19 −2
  • docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md +9 −1
  • instances_cli.go +2 −2
  • internal/activator/activator_test.go +2 −2
  • internal/console/apiinstances.go +24 −2
  • internal/console/apiinstances_test.go +49 −0
  • internal/console/apirelay.go +19 −0
  • internal/console/apirelay_test.go +43 −1
  • internal/console/apiroutes.go +6 −1
  • internal/console/apistore.go +66 −0
  • internal/console/apistore_test.go +60 −0
  • internal/console/relayops.go +8 −1
  • internal/host/agent.go +3 −0
  • internal/host/metrics_probes_test.go +26 −0
  • internal/host/probes.go +20 −0
  • internal/host/schedule_rearm_wake_test.go +119 −0
  • internal/registry/registry.go +3 −0
  • internal/registry/timers.go +35 −24
  • internal/registry/timers_test.go +169 −3
  • internal/relay/authority.go +5 −0
  • internal/relay/masters.go +19 −0
  • internal/relay/masters_test.go +23 −0
  • internal/relay/race_off_test.go +6 −0
  • internal/relay/race_on_test.go +12 −0
  • internal/relay/serve.go +35 −14
  • internal/relay/serve_test.go +192 −1
  • internal/relay/server_test.go +37 −0
  • internal/schedule/control.go +12 −1
  • internal/schedule/control_test.go +63 −0
  • internal/schedule/converge_test.go +12 −3
  • internal/schedule/runner.go +21 −6
  • internal/schedule/runner_test.go +307 −3
  • internal/status/snapshot.go +29 −2
  • internal/status/status_test.go +55 −0
  • relay.go +18 −0
  • relay_test.go +19 −0
  • routes_cli.go +13 −4
  • store.go +16 −0
  • store_test.go +26 −0

CI passed — run details

diff --git a/apimode.go b/apimode.go
index 5b7b0cec7..7e8d0cd6f 100644
--- a/apimode.go
+++ b/apimode.go
@@ -1657,6 +1657,17 @@ type apiStoreReply struct {
// exactly as it always did.
Scanning *apiScanBlock `json:"scanning,omitempty"`
Entitlement *objstore.Entitlement `json:"entitlement"`
+ // Size is the bucket's size from the console's daily storage sweep, nil
+ // before a sweep has measured it (and from an older console).
+ Size *apiStoreSize `json:"size"`
+}
+
+// apiStoreSize mirrors the console's storeSizeReply (internal/console/apistore.go).
+type apiStoreSize struct {
+ AsOf string `json:"as_of"`
+ StoreBytes int64 `json:"store_bytes"`
+ Unswept bool `json:"unswept,omitempty"`
+ BucketBytes int64 `json:"bucket_bytes"`
}
const storeProvisionHTTPTimeout = 75 * time.Second
@@ -2098,7 +2109,9 @@ type apiRelayRegion struct {
LeaseLive bool `json:"lease_live"`
Listeners []string `json:"listeners"`
Kids []string `json:"kids"`
- Stale bool `json:"stale"`
+ // kid -> RFC3339 retire_at, for the loaded masters that carry one.
+ Retiring map[string]string `json:"retiring,omitempty"`
+ Stale bool `json:"stale"`
// The relay's own last two errors: why it stopped reading. Printed on a
// stale row, because the state without the reason is not actionable.
MastersError string `json:"masters_error,omitempty"`
@@ -2857,6 +2870,10 @@ type apiInstanceRecord struct {
// column can give, since the person reading it is checking whether
// something private is exposed.
Gated *bool `json:"gated"`
+ // DBBytes is the instance's SQLite database size as its box last
+ // measured it (instanceListRecord.DBBytes). nil when unknown, or from a
+ // console that predates the field — the DB column prints a dash.
+ DBBytes *int64 `json:"db_bytes,omitempty"`
}
// apiUpstreamState mirrors status.UpstreamState's wire shape (internal/status)
@@ -3122,6 +3139,7 @@ type apiRouteRow struct {
Backing string `json:"backing"`
Provisioned bool `json:"provisioned"`
Replication string `json:"replication"`
+ DBBytes *int64 `json:"db_bytes,omitempty"` // nil = unknown, printed as a dash
}
// apiRoutePathRow is one path route of the same answer: a leading URL
diff --git a/docs/site/cli.md b/docs/site/cli.md
index 8abcac2a7..60d098f31 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -560,6 +560,12 @@ carlos instances list --app hello
`not-responding` and the rest. That is usually what you want when something
is wrong.
+The `DB` column is the size of each instance's SQLite database on disk,
+including its write-ahead log, as its box last measured it. Boxes measure
+every few minutes, and asleep instances are measured too. A dash means no box
+has measured it: the instance is not running anywhere yet, it has not created
+its database, or it keeps its data somewhere CARLOS does not manage.
+
`--environment <e>` on `create`, and `set-environment` on a route that
already exists, bind one instance to one of the app's named config bundles.
This matters most for a route standing **alongside** a live one. Config is
@@ -670,6 +676,9 @@ carlos routes --app hello
It reads the console's own box, so on a multi-box fleet a short answer is not
necessarily a complete one. The command says so every time it prints.
+Its `DB` column is each route's SQLite database size, the same figure as in
+[carlos instances](/docs/cli#carlos-instances).
+
### carlos domains
Attaches a customer's own hostname to one of the app's routes, detaches it,
@@ -822,6 +831,12 @@ after the cloud provider destroys it; during that destroy/re-create window,
retry after the provider releases the name. `status` shows an incomplete store
so you can tell this from an undeclared store.
+Once the store is provisioned, `status` also shows how big the bucket is. That
+figure comes from a sweep the console runs once a day, so it carries the date
+it was measured and can be up to a day old. Sizing a bucket means listing
+every object in it, and `status` does not do that on each call. A fresh store
+reads "not measured yet" until the next sweep.
+
`carlos store rotate` mints a fresh credential and leaves the old one live
until you run it again with `--finish`, so the app has a window to pick the
new one up. Rotating, declaring and reading status are member verbs and want
@@ -934,8 +949,10 @@ app holds ends on the relays' next authority read, within a minute.
TCP-address route it restarted by hand now runs the new key — `status` names
the routes waiting on that, since they are the one shape this platform
cannot restart for you. `rotate --region <r>` is the operator's verb after
-staging a new master; `keygen` prints one. `serve` is the unit's entry point
-and not a thing to run by hand.
+staging a new master; `keygen` prints one. After a rotation, `status` names
+any master that is retiring and the time it may be dropped, box by box, since
+a region is done with a key only once every relay there has stopped loading
+it. `serve` is the unit's entry point and not a thing to run by hand.
`keygen` prints one line — a key id, a key, and `-` for "no retirement date"
— for a region's master key. Stage that line in `/etc/carlos/relay-masters`
diff --git a/docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md b/docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md
index bdc935a46..087801786 100644
--- a/docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md
+++ b/docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md
@@ -393,7 +393,15 @@ ours. **(codex R1-20)**
- `apps/<acct>/<app>/relay.json`, CAS-guarded like `store.json`, is the
**source of truth** for the app: `{enabled, incarnation, regions,
- op, phase, config_generation, enabled_at, disabled_at}`. `op` is a
+ op, phase, issued_digest, issued_suffixes, cutovers, enabled_at,
+ disabled_at}`. `issued_digest` is the sha256 of the sealed envelope
+ that was delivered, which is what `Record.IssuedDigest` records and
+ what reconcile compares; an earlier draft of this spec called it
+ `config_generation` and no implementation ever carried that name.
+ `issued_suffixes` (region to the username suffix last issued) and
+ `cutovers` (the per-route member acknowledgements) are the other two
+ fields `relay.Record` has always carried and this listing has always
+ omitted. **(discussion #120)** `op` is a
counter bumped by every operation; `phase` is one of `config`,
`aggregate`, `done` and records how far the operation got. The
incarnation is sixteen hex characters drawn at random at first
diff --git a/instances_cli.go b/instances_cli.go
index ad4fdc5f9..5ffc16865 100644
--- a/instances_cli.go
+++ b/instances_cli.go
@@ -370,13 +370,13 @@ func runInstancesList(args []string) {
log.Fatalf("instances list: %v", err)
}
w := newTable(stdout)
- fmt.Fprintln(w, "HOST\tREGION\tCHANNEL\tPOOL\tVERSION\tHEALTH\tUPSTREAMS\tCREATED\tBY")
+ fmt.Fprintln(w, "HOST\tREGION\tCHANNEL\tPOOL\tVERSION\tHEALTH\tUPSTREAMS\tDB\tCREATED\tBY")
for _, r := range recs {
rowHealth := r.Health
if rowHealth == "" {
rowHealth = "—" // an old console: the field predates it, not "unhealthy"
}
- fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", r.Host, r.Region, r.Channel, r.Pool, r.StagedVersion, rowHealth, upstreamsCell(r.Upstreams), r.CreatedAt, r.CreatedBy)
+ fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", r.Host, r.Region, r.Channel, r.Pool, r.StagedVersion, rowHealth, upstreamsCell(r.Upstreams), humanBytes(r.DBBytes), r.CreatedAt, r.CreatedBy)
}
w.Flush()
if truncated {
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index e2cc159aa..9039e321c 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -4753,8 +4753,8 @@ func stageRecordingLiveFullEnv(t *testing.T, b *ExecBackend, host, ver string) (
argvPath = filepath.Join(dir, "argv")
envPath = filepath.Join(dir, "env")
script := "#!/bin/sh\n" +
- "printf '%s\\n' \"$@\" > '" + argvPath + "'\n" +
- "env > '" + envPath + "'\n"
+ "printf '%s\\n' \"$@\" > '" + argvPath + ".tmp' && mv '" + argvPath + ".tmp' '" + argvPath + "'\n" +
+ "env > '" + envPath + ".tmp' && mv '" + envPath + ".tmp' '" + envPath + "'\n"
vdir := filepath.Join(b.OptDir, "versions", ver)
if err := os.MkdirAll(vdir, 0o755); err != nil {
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go
index b85fe8b8e..1256c0024 100644
--- a/internal/console/apiinstances.go
+++ b/internal/console/apiinstances.go
@@ -184,6 +184,12 @@ type instanceListRecord struct {
// is always "" and omitted. Gated is what a member may know.
HiddenGate string `json:"gate,omitempty"`
Gated bool `json:"gated"`
+ // DBBytes is the on-disk size of this instance's SQLite database (file
+ // plus -wal) as its box last measured it (status.BoxMetrics.RouteDBBytes),
+ // under the same staleness rule as StagedVersion. Absent when no fresh
+ // box has sized it: not minted yet, no database file yet, or a route
+ // whose database lives somewhere the platform does not know about.
+ DBBytes *int64 `json:"db_bytes,omitempty"`
}
// instanceHealth computes each record host's health word the same way the
@@ -321,6 +327,18 @@ func (s *Server) healthByHost(ctx context.Context, routeFor map[string]registry.
type boxJoin struct {
versions map[string]string
upstreams map[string][]status.UpstreamState
+ dbBytes map[string]int64
+}
+
+// dbSize is the join's database size for host, nil when no fresh box
+// reported one — so the wire field is absent rather than a 0 that reads as
+// an empty database.
+func (j boxJoin) dbSize(host string) *int64 {
+ n, ok := j.dbBytes[host]
+ if !ok {
+ return nil
+ }
+ return &n
}
// loadBoxJoin builds boxJoin from every NON-STALE box's own
@@ -356,7 +374,8 @@ func loadBoxJoin(ctx context.Context, s release.Store, now time.Time) boxJoin {
log.Printf("console: api instances: could not load status boxes for the list join: %v", err)
return boxJoin{}
}
- j := boxJoin{versions: map[string]string{}, upstreams: map[string][]status.UpstreamState{}}
+ j := boxJoin{versions: map[string]string{}, upstreams: map[string][]status.UpstreamState{},
+ dbBytes: map[string]int64{}}
for _, b := range boxes {
if b.Metrics == nil || status.BoxStale(b, now, status.SnapshotPeriod) {
continue
@@ -367,6 +386,9 @@ func loadBoxJoin(ctx context.Context, s release.Store, now time.Time) boxJoin {
for host, ups := range b.Metrics.Upstreams {
j.upstreams[host] = ups
}
+ for host, n := range b.Metrics.RouteDBBytes {
+ j.dbBytes[host] = n
+ }
}
return j
}
@@ -428,7 +450,7 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) {
}
matched = append(matched, instanceListRecord{Record: rec,
StagedVersion: join.versions[rec.Host], Health: h, Upstreams: join.upstreams[rec.Host],
- ServedBy: served[rec.Host], Gated: rec.Gate != ""})
+ ServedBy: served[rec.Host], Gated: rec.Gate != "", DBBytes: join.dbSize(rec.Host)})
}
total := len(matched)
// provision.ListRecords returns records host-sorted and nothing above
diff --git a/internal/console/apiinstances_test.go b/internal/console/apiinstances_test.go
index 6c767d9fb..2c02e9063 100644
--- a/internal/console/apiinstances_test.go
+++ b/internal/console/apiinstances_test.go
@@ -2556,3 +2556,52 @@ func TestAPIInstanceUpdateGateRefusals(t *testing.T) {
t.Fatalf("empty-body refusal should name gate: %s", out)
}
}
+
+// TestAPIInstanceListDBBytes: db_bytes joins from a FRESH box's
+// Metrics.RouteDBBytes, is absent (not 0) for a host no box sized, and is
+// dropped for a host whose only reporting box has gone stale — the same
+// rules as staged_version.
+func TestAPIInstanceListDBBytes(t *testing.T) {
+ fx, token := instanceFixture(t, "code-inst-dbbytes")
+ enableInstances(t, fx, "titogo", titoManifest())
+ u := instancesURL(fx, "bab", "titogo")
+ for _, h := range []string{"acme.go.tito.io", "beta.go.tito.io", "gamma.go.tito.io"} {
+ if code, _, out := rawReq(t, http.MethodPost, u, token,
+ fmt.Sprintf(`{"host":%q,"region":"ie"}`, h)); code != http.StatusCreated {
+ t.Fatalf("create %s = %d %s", h, code, out)
+ }
+ }
+ fresh := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now())
+ fresh.Metrics = &status.BoxMetrics{RouteDBBytes: map[string]int64{"acme.go.tito.io": 123456}}
+ if err := status.WriteBox(context.Background(), fx.rel, fresh); err != nil {
+ t.Fatal(err)
+ }
+ stale := status.CollectBox(nil, status.Probes{MachineID: "box2"}, time.Now().Add(-time.Hour))
+ stale.Metrics = &status.BoxMetrics{RouteDBBytes: map[string]int64{"gamma.go.tito.io": 999}}
+ if err := status.WriteBox(context.Background(), fx.rel, stale); err != nil {
+ t.Fatal(err)
+ }
+
+ code, _, out := rawReq(t, http.MethodGet, u, token, "")
+ if code != http.StatusOK {
+ t.Fatalf("list = %d %s", code, out)
+ }
+ var list struct {
+ Instances []map[string]any `json:"instances"`
+ }
+ if err := json.Unmarshal([]byte(out), &list); err != nil {
+ t.Fatalf("list body %s: %v", out, err)
+ }
+ got := map[string]map[string]any{}
+ for _, r := range list.Instances {
+ got[r["host"].(string)] = r
+ }
+ if v := got["acme.go.tito.io"]["db_bytes"]; v != float64(123456) {
+ t.Errorf("acme db_bytes = %v, want 123456", v)
+ }
+ for _, h := range []string{"beta.go.tito.io", "gamma.go.tito.io"} {
+ if v, ok := got[h]["db_bytes"]; ok {
+ t.Errorf("%s db_bytes = %v, want the key absent", h, v)
+ }
+ }
+}
diff --git a/internal/console/apirelay.go b/internal/console/apirelay.go
index 18d8c8fda..b2d40fbb0 100644
--- a/internal/console/apirelay.go
+++ b/internal/console/apirelay.go
@@ -171,6 +171,13 @@ type apiRelayRegion struct {
LeaseLive bool `json:"lease_live"` // relay.Lease still armed to serve (15-minute ceiling)
Listeners []string `json:"listeners"`
Kids []string `json:"kids"`
+ // Retiring is kid -> RFC3339 retire_at for the masters this box loaded
+ // that carry one. It answers the operator's mid-rotation question —
+ // when may the old key leave /etc/carlos/relay-masters — from the CLI,
+ // per box, which is the only place the answer is true: a region is only
+ // done with a key once EVERY relay there has stopped loading it
+ // (discussion #120).
+ Retiring map[string]string `json:"retiring,omitempty"`
// Stale mirrors relayRotate's own gate: a heartbeat whose last successful
// authority read is older than relayHeartbeatFresh (five minutes) cannot
// be read as current, so a dead relay cannot look alive by silence.
@@ -214,6 +221,12 @@ type apiRelayReply struct {
// retirement date on an app's status would mean either inventing state the
// console does not keep or recomputing it from a clock, which would tell a
// member a retirement deadline that drifts every time they ask.
+ //
+ // apiRelayRegion.Retiring is NOT that value and is not a softening of this
+ // rule (discussion #120). It is the retire_at a relay actually loaded out
+ // of its own masters file, reported per box and echoed verbatim: state the
+ // operator wrote, not state this console computed, so it cannot drift
+ // between two reads.
}
// apiRelayRotateReply is what `carlos relay rotate` answers with — a region
@@ -461,6 +474,12 @@ func (s *Server) relayReply(ctx context.Context, acct *Account, app string, rec
MastersError: hb.MastersError,
AuthorityError: hb.AuthorityError,
}
+ for kid, at := range hb.Retiring {
+ if row.Retiring == nil {
+ row.Retiring = map[string]string{}
+ }
+ row.Retiring[kid] = at.UTC().Format(time.RFC3339)
+ }
if hb.ReadAt.IsZero() {
row.HeartbeatAge = "never"
} else {
diff --git a/internal/console/apirelay_test.go b/internal/console/apirelay_test.go
index f54ab92c5..b5762f374 100644
--- a/internal/console/apirelay_test.go
+++ b/internal/console/apirelay_test.go
@@ -36,11 +36,18 @@ func relayURL(fx *consoleFixture, app string) string {
// with its last SUCCESSFUL authority read `age` in the past — the clock
// status reads, and the one a dead relay cannot move.
func seedHeartbeat(t *testing.T, fx *consoleFixture, box, region string, age time.Duration, kids ...string) {
+ t.Helper()
+ seedHeartbeatRetiring(t, fx, box, region, age, nil, kids...)
+}
+
+// seedHeartbeatRetiring is seedHeartbeat with the retiring map set, for the
+// one test that covers the hop from a relay's heartbeat to the status reply.
+func seedHeartbeatRetiring(t *testing.T, fx *consoleFixture, box, region string, age time.Duration, retiring map[string]time.Time, kids ...string) {
t.Helper()
now := time.Now()
hb := relay.Heartbeat{
Box: box, Region: region, ReadAt: now.Add(-age), WrittenAt: now.Add(-age),
- Listeners: []string{"udp:3478", "tcp:3478"}, Kids: kids,
+ Listeners: []string{"udp:3478", "tcp:3478"}, Kids: kids, Retiring: retiring,
}
if err := relay.WriteHeartbeat(context.Background(), fx.rel, hb); err != nil {
t.Fatalf("seed heartbeat: %v", err)
@@ -741,3 +748,38 @@ func TestAPIRelayStatusNeverAdoptsOnEmptyIssuedSuffixes(t *testing.T) {
t.Fatalf("delivery = %q, an empty-IssuedSuffixes record must never read as adopted", got)
}
}
+
+// TestAPIRelayStatusCarriesRetiringMasters covers the hop from a relay's
+// heartbeat to the status reply (discussion #120). Without it the console
+// half of the retiring display can be deleted with the whole suite green,
+// and an operator mid-rotation silently reads "nothing is retiring" — which
+// is indistinguishable from the true answer and is the wrong one to act on.
+func TestAPIRelayStatusCarriesRetiringMasters(t *testing.T) {
+ fx, _ := relayFixture(t, "code-relay-retiring")
+ token := cliToken(t, fx, "alice@keymail.dev", "code-relay-retiring-2")
+ if status, out := postJSON(t, relayURL(fx, "titogo"), map[string]any{}, token); status != http.StatusOK {
+ t.Fatalf("enable = %d %v, want 200", status, out)
+ }
+
+ retire := time.Date(2026, 9, 9, 12, 0, 0, 0, time.UTC)
+ seedHeartbeatRetiring(t, fx, "box1", "eu-west-1", time.Minute,
+ map[string]time.Time{"k1": retire}, "k1", "k2")
+
+ status, out := getJSON(t, relayURL(fx, "titogo"), token)
+ if status != http.StatusOK {
+ t.Fatalf("status = %d %v, want 200", status, out)
+ }
+ row := regionRow(t, out, 0)
+ got, ok := row["retiring"].(map[string]any)
+ if !ok {
+ t.Fatalf("region row carries no retiring map: %#v", row)
+ }
+ if got["k1"] != retire.Format(time.RFC3339) {
+ t.Errorf("retiring[k1] = %v, want %s", got["k1"], retire.Format(time.RFC3339))
+ }
+ // k2 is loaded but carries no retirement date, so it must be absent
+ // rather than present with a zero time.
+ if _, ok := got["k2"]; ok {
+ t.Errorf("k2 has no retirement date and must not appear: %#v", got)
+ }
+}
diff --git a/internal/console/apiroutes.go b/internal/console/apiroutes.go
index bf16ca8dc..5dcd3f5ad 100644
--- a/internal/console/apiroutes.go
+++ b/internal/console/apiroutes.go
@@ -67,6 +67,10 @@ type apiRoute struct {
// absent field as "", which is exactly what every route carried before
// there was a door onto the column (discussion #10).
Replication string `json:"replication,omitempty"`
+ // DBBytes is the route's SQLite database size as its box last reported
+ // it — the same box-snapshot join, and the same staleness rule, as the
+ // instance list's db_bytes. Absent when unknown, never 0.
+ DBBytes *int64 `json:"db_bytes,omitempty"`
}
// apiRoutePath is one path route: a leading URL segment on Host that is
@@ -144,6 +148,7 @@ func (s *Server) handleAPIRoutes(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), routeListTimeout)
defer cancel()
health := s.healthByHost(ctx, routeFor)
+ join := loadBoxJoin(ctx, s.cfg.Releases, time.Now())
rows := make([]apiRoute, 0, len(routeFor))
for _, rt := range routeFor {
rows = append(rows, apiRoute{
@@ -151,7 +156,7 @@ func (s *Server) handleAPIRoutes(w http.ResponseWriter, r *http.Request) {
Channel: rt.Channel, Environment: rt.Environment, Status: rt.Status,
Health: health[rt.Host], Hibernate: rt.Hibernate, Unit: rt.Unit,
Backing: rt.Backing, Provisioned: rt.Provisioned,
- Replication: rt.Replication,
+ Replication: rt.Replication, DBBytes: join.dbSize(rt.Host),
})
}
// Sorted by host: registry order is not stable across passes, and a
diff --git a/internal/console/apistore.go b/internal/console/apistore.go
index aeb419075..1db526b03 100644
--- a/internal/console/apistore.go
+++ b/internal/console/apistore.go
@@ -35,16 +35,19 @@ package console
import (
"context"
"crypto/ecdh"
+ "encoding/json"
"errors"
"fmt"
"log"
"net/http"
+ "os"
"strings"
"time"
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
)
// storeTimeout bounds every store round trip these handlers make — the same
@@ -221,9 +224,72 @@ func (s *Server) storeReply(ctx context.Context, acct *Account, app string, rec
// would be a half-truth — but it is nil on every store that never
// opted in, which is all of them by default.
"scanning": rec.Scanning,
+ // The bucket's size from the daily storage sweep, absent until a
+ // sweep has measured this store. See storeSize.
+ "size": s.storeSize(ctx, acct, app, rec),
}, nil
}
+// storeSizeLookback is how many UTC days storeSize walks back looking for a
+// storage sweep. The sweep runs on the console's first hourly economics wake
+// of each day, so today or yesterday normally answers; a week covers a
+// console that was down for a while, and anything older is too stale to
+// print as the bucket's size.
+const storeSizeLookback = 7
+
+// storeSizeReply is the "size" block of `carlos store status`. It is copied
+// from the storage sweep (usage/storage/<date>.json) and is NOT measured on
+// this request: sizing a bucket means listing every object in it, which is
+// what the daily sweep already does once. Unswept means the sweep ran but
+// could not size this store (bad credential, endpoint down), and then
+// StoreBytes is meaningless. BucketBytes is the app's own files in the
+// deployment bucket (releases, config), a separate figure from its store.
+type storeSizeReply struct {
+ AsOf string `json:"as_of"`
+ StoreBytes int64 `json:"store_bytes"`
+ Unswept bool `json:"unswept,omitempty"`
+ BucketBytes int64 `json:"bucket_bytes"`
+}
+
+// storeSize finds the newest storage sweep from the last storeSizeLookback
+// days that has an answer for this app's store, or nil. A sweep that did not
+// see the store provisioned (it ran before the grant, or predates the
+// store_provisioned field) has no answer: its StoreBytes of 0 would read as
+// an empty bucket. Read failures are logged and answer nil; the size is an
+// annotation on status, never a reason for status to fail.
+func (s *Server) storeSize(ctx context.Context, acct *Account, app string, rec *objstore.Record) *storeSizeReply {
+ if rec.Provisioned == nil {
+ return nil
+ }
+ key := acct.Sqid + "/" + app
+ now := time.Now()
+ for i := 0; i < storeSizeLookback; i++ {
+ date := usage.DayOf(now.AddDate(0, 0, -i))
+ raw, _, err := s.cfg.Releases.Get(ctx, usage.StorageDayKey(date))
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ log.Printf("console: api store size %s: %v", usage.StorageDayKey(date), err)
+ return nil
+ }
+ var day usage.StorageDay
+ if err := json.Unmarshal(raw, &day); err != nil {
+ log.Printf("console: api store size: parse %s: %v", usage.StorageDayKey(date), err)
+ return nil
+ }
+ st, ok := day.Apps[key]
+ if !ok || (!st.StoreProvisioned && !st.StoreUnswept) {
+ // The newest sweep did not see this store; an older one cannot
+ // have either.
+ return nil
+ }
+ return &storeSizeReply{AsOf: date, StoreBytes: st.StoreBytes,
+ Unswept: st.StoreUnswept, BucketBytes: st.BucketBytes}
+ }
+ return nil
+}
+
// handleAPIStoreCreate answers `carlos store create`: record that this app
// wants object storage, under a chosen (or default) env prefix. It is a
// REQUEST, not a provisioning — nothing exists in the backend when this
diff --git a/internal/console/apistore_test.go b/internal/console/apistore_test.go
index 02c128e6e..d0a91832f 100644
--- a/internal/console/apistore_test.go
+++ b/internal/console/apistore_test.go
@@ -11,6 +11,7 @@ package console
import (
"context"
+ "encoding/json"
"errors"
"net/http"
"strings"
@@ -20,6 +21,7 @@ import (
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
)
func storeFixture(t *testing.T, code string) (*consoleFixture, string) {
@@ -1136,3 +1138,61 @@ func TestAPIStoreGrantFailureSaysResume(t *testing.T) {
t.Fatalf("buckets after the resume = %v", fake.Buckets)
}
}
+
+// TestAPIStoreStatusCarriesTheSweptSize: `size` is copied from the newest
+// storage sweep, dated. It is null when no sweep has measured the store, and
+// a sweep that did not see the store provisioned is never shown (its 0 would
+// read as an empty bucket).
+func TestAPIStoreStatusCarriesTheSweptSize(t *testing.T) {
+ fx, token := storeFixture(t, "code-store-size")
+ ctx := context.Background()
+ now := time.Now()
+ putRecord(t, fx, "titogo", &objstore.Record{
+ Declared: &objstore.Declared{EnvPrefix: "BLOBS", At: now, By: "alice@keymail.dev"},
+ Grant: &objstore.GrantBlock{Backend: "aws", At: now, By: "paul@keymail.dev"},
+ Provisioned: &objstore.Provisioned{Bucket: "carlos-bab-titogo", Region: "eu-west-1",
+ IAMUser: "u", KeyID: "AKIA", At: now},
+ })
+ key := release.OpsAccount + "/titogo"
+ seed := func(day time.Time, st usage.AppStorage) {
+ t.Helper()
+ body, _ := json.Marshal(usage.StorageDay{V: 1, Date: usage.DayOf(day),
+ Apps: map[string]usage.AppStorage{key: st}})
+ if _, err := fx.rel.Put(ctx, usage.StorageDayKey(usage.DayOf(day)), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ size := func() map[string]any {
+ t.Helper()
+ code, out := getJSON(t, storeURL(fx, "titogo"), token)
+ if code != http.StatusOK {
+ t.Fatalf("status = %d %v", code, out)
+ }
+ if out["size"] == nil {
+ return nil
+ }
+ return block(t, out, "size")
+ }
+
+ if got := size(); got != nil {
+ t.Fatalf("size = %v with no sweep at all, want null", got)
+ }
+
+ // A sweep that ran before the grant: the store was not there to size.
+ seed(now.AddDate(0, 0, -2), usage.AppStorage{BucketBytes: 10})
+ if got := size(); got != nil {
+ t.Fatalf("size = %v from a sweep that did not see the store, want null", got)
+ }
+
+ seed(now.AddDate(0, 0, -1), usage.AppStorage{BucketBytes: 10, StoreBytes: 5 << 20, StoreProvisioned: true})
+ got := size()
+ if got == nil || got["store_bytes"] != float64(5<<20) || got["as_of"] != usage.DayOf(now.AddDate(0, 0, -1)) {
+ t.Fatalf("size = %v, want yesterday's 5MB", got)
+ }
+
+ // Today's sweep could not read the bucket: the newest answer says so.
+ seed(now, usage.AppStorage{BucketBytes: 10, StoreUnswept: true})
+ if got := size(); got == nil || got["unswept"] != true || got["as_of"] != usage.DayOf(now) {
+ t.Fatalf("size = %v, want today's unswept marker", got)
+ }
+}
diff --git a/internal/console/relayops.go b/internal/console/relayops.go
index 3c26d297f..719e196df 100644
--- a/internal/console/relayops.go
+++ b/internal/console/relayops.go
@@ -123,8 +123,15 @@ var (
"this deployment runs no relay (CARLOS_RELAY_REGIONS is empty) — ask the operator to arm a region before enabling it for an app")
errRelayNoSealKey = errors.New(
"this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a relay key, so it will not issue one")
+ // The commonest cause is not another console at all: it is THIS console
+ // having just restarted, leaving its own previous process's lease to
+ // expire. That takes up to relayLeaseTTL, so the message says to retry
+ // and names the wait rather than leaving an operator to guess whether
+ // the deployment is wedged (discussion #120).
errRelayNotHeld = errors.New(
- "this console does not hold the relay writer lease — another console is driving the relay; the operation was not started")
+ "this console does not hold the relay writer lease. another console is driving the relay, " +
+ "or this one restarted and its previous lease has not expired yet. the operation was not started, " +
+ "so retry in up to a minute")
errRelayNotEnabled = errors.New("the relay is not enabled for this app")
errRelayUndelivered = errors.New("a route has no config delivery")
// errRelayUnknownHost is `carlos relay adopted --host` naming something
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 2b7f32e8a..af51537a3 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -2705,6 +2705,9 @@ func statusLoop(reg *registry.Registry) {
// edge-ready callback publishes the source, and on any box that
// probes nothing — both publish no key at all.
Upstreams: upstreamStates(),
+ // One stat (two with a WAL) per database route per tick, which
+ // is what `carlos instances list` and `carlos routes` show as DB.
+ DBBytes: dbFileBytes,
}
// A fleet box (CARLOS_FLEET set) never gets here as a contender: it
diff --git a/internal/host/metrics_probes_test.go b/internal/host/metrics_probes_test.go
index adae5520e..65100a32a 100644
--- a/internal/host/metrics_probes_test.go
+++ b/internal/host/metrics_probes_test.go
@@ -237,3 +237,29 @@ func TestListCarlosUnitsFindsUnitsUnderCarlosSlice(t *testing.T) {
t.Fatalf("carlos-edge.service missing from the combined scan: %+v", combined)
}
}
+
+// TestDBFileBytes: the main file plus its -wal, nil (unknown) for a database
+// that does not exist, and the main file alone once a checkpoint has removed
+// the WAL.
+func TestDBFileBytes(t *testing.T) {
+ dir := t.TempDir()
+ db := filepath.Join(dir, "h.test.db")
+ if got := dbFileBytes(db); got != nil {
+ t.Fatalf("missing db = %d, want nil", *got)
+ }
+ if err := os.WriteFile(db, make([]byte, 4096), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if got := dbFileBytes(db); got == nil || *got != 4096 {
+ t.Fatalf("db alone = %v, want 4096", got)
+ }
+ if err := os.WriteFile(db+"-wal", make([]byte, 1000), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if got := dbFileBytes(db); got == nil || *got != 5096 {
+ t.Fatalf("db + wal = %v, want 5096", got)
+ }
+ if got := dbFileBytes(dir); got != nil {
+ t.Fatalf("a directory = %d, want nil", *got)
+ }
+}
diff --git a/internal/host/probes.go b/internal/host/probes.go
index c17681403..d9e889387 100644
--- a/internal/host/probes.go
+++ b/internal/host/probes.go
@@ -376,3 +376,23 @@ func readDiskFreeBytes() *int64 {
freeBytes := int64(avail)
return &freeBytes
}
+
+// dbFileBytes is status.Probes.DBBytes: the on-disk size of a route's SQLite
+// database, the main file plus its -wal. The WAL counts because in WAL mode
+// recent writes live there until a checkpoint, so the main file alone
+// understates a busy database. A missing or unreadable main file is nil
+// (unknown), never 0: a route that has not created its database yet has no
+// size. A missing -wal is normal after a checkpoint and adds nothing. A
+// hibernating route's file stays on disk while it sleeps (the activator keeps
+// it as a warm cache), so asleep routes are sized too.
+func dbFileBytes(dbPath string) *int64 {
+ fi, err := os.Stat(dbPath)
+ if err != nil || !fi.Mode().IsRegular() {
+ return nil
+ }
+ n := fi.Size()
+ if wal, err := os.Stat(dbPath + "-wal"); err == nil && wal.Mode().IsRegular() {
+ n += wal.Size()
+ }
+ return &n
+}
diff --git a/internal/host/schedule_rearm_wake_test.go b/internal/host/schedule_rearm_wake_test.go
new file mode 100644
index 000000000..560861d41
--- /dev/null
+++ b/internal/host/schedule_rearm_wake_test.go
@@ -0,0 +1,119 @@
+package host
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+func TestSleepingInstanceRearmsItsNextDeliveryBeforeReturning(t *testing.T) {
+ const host = "rearm.bab.oncarlos.com"
+ const idle = 200 * time.Millisecond
+ entered := make(chan struct{}, 1)
+ release := make(chan struct{})
+ events := make(chan string, 4)
+ var calls atomic.Int32
+ var reg *registry.Registry
+ now := time.Now().UTC()
+ var clock atomic.Int64
+ clock.Store(now.UnixNano())
+ next := now.Add(time.Hour).Unix()
+ var firstError atomic.Pointer[error]
+ var instReg *registry.Registry
+ instReg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, req *http.Request) {
+ if req.Method != http.MethodPost || req.URL.Path != "/jobs/reminders" || !strings.HasPrefix(req.Header.Get("Authorization"), "Bearer ") {
+ w.WriteHeader(http.StatusForbidden)
+ return
+ }
+ if calls.Add(1) == 1 {
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "next-due", Kind: schedule.KindOnce,
+ Spec: "at " + strconv.FormatInt(next, 10), Path: "/jobs/reminders", NextAt: next, DueAt: next}); err != nil {
+ firstError.Store(&err)
+ }
+ entered <- struct{}{}
+ <-release
+ }
+ w.WriteHeader(http.StatusNoContent)
+ })
+ reg = instReg
+ t.Cleanup(func() {
+ select {
+ case <-release:
+ default:
+ close(release)
+ }
+ })
+ runner := &schedule.Runner{Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
+ Now: func() time.Time { return time.Unix(0, clock.Load()) }, Timeout: 10 * time.Second,
+ Event: func(_ registry.Route, _, msg string) { events <- msg },
+ }
+ due := now.Unix() - 1
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: "next-due", Kind: schedule.KindOnce,
+ Spec: fmt.Sprintf("at %d", due), Path: "/jobs/reminders", NextAt: due, DueAt: due}); err != nil {
+ t.Fatal(err)
+ }
+ runner.Tick(context.Background())
+ select {
+ case <-entered:
+ case <-time.After(10 * time.Second):
+ t.Fatal("sleeping instance never received its due delivery")
+ }
+ if err := firstError.Load(); err != nil {
+ t.Fatal(*err)
+ }
+ time.Sleep(3 * idle)
+ if !inst.running() || !svc.Owns(host) {
+ t.Fatal("instance slept while its scheduled handler was still running")
+ }
+ close(release)
+ select {
+ case <-events:
+ case <-time.After(10 * time.Second):
+ t.Fatal("first delivery did not finish")
+ }
+ timer, err := reg.GetTimer(host, "next-due")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if timer.NextAt != next {
+ t.Fatalf("completion erased the replacement delivery: next_at=%d, want %d", timer.NextAt, next)
+ }
+ deadline := time.Now().Add(10 * time.Second)
+ for inst.running() || svc.Owns(host) {
+ if time.Now().After(deadline) {
+ t.Fatal("instance did not hibernate between deliveries")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ clock.Store(time.Unix(next, 0).UnixNano())
+ for calls.Load() < 2 {
+ runner.Tick(context.Background())
+ if time.Now().After(deadline) {
+ t.Fatal("replacement delivery never woke the instance")
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ select {
+ case <-events:
+ case <-time.After(10 * time.Second):
+ t.Fatal("second delivery did not finish")
+ }
+ timer, err = reg.GetTimer(host, "next-due")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if timer.NextAt != 0 || timer.LastStatus != "204" {
+ t.Fatalf("second delivery not completed: %+v", timer)
+ }
+ if inst.startCount() != 2 {
+ t.Fatalf("got %d wakes, want 2", inst.startCount())
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 7dfc050cd..2d1f935aa 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -550,6 +550,9 @@ CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`);
if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN due_at INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
}
+ if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN registration TEXT NOT NULL DEFAULT ''`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
+ return err
+ }
if err := r.migrateAccounts(); err != nil {
return err
}
diff --git a/internal/registry/timers.go b/internal/registry/timers.go
index 12cdde238..cec61318c 100644
--- a/internal/registry/timers.go
+++ b/internal/registry/timers.go
@@ -28,41 +28,43 @@ type Timer struct {
// when the row advances to a new occurrence (MarkTimerFired with a new
// due, FireTimerNow, or an upsert that moves next_at). Zero on rows
// written before the column existed; readers fall back to NextAt.
- DueAt int64
+ DueAt int64
+ Registration string `json:"-"`
}
var ErrTimerKind = errors.New("registry: timer exists with a different kind")
-const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at`
+const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration`
func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
var t Timer
- err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt)
+ err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt, &t.Registration)
return t, err
}
-// UpsertTimer inserts t, or updates an existing row's spec/path — moving
-// next_at (and due_at with it) only when spec or path changed or t.NextAt
-// is non-zero — and never touches the runner's columns on update.
+// UpsertTimer inserts t, or updates an existing row's spec/path. A changed spec/path or non-zero NextAt starts a new registration and resets attempts; unchanged declarations preserve delivery state.
func (r *Registry) UpsertTimer(t Timer) error {
- existing, err := r.GetTimer(t.Host, t.Name)
- switch {
- case errors.Is(err, os.ErrNotExist):
- _, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`,
- t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
+ const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
+ res, err := r.sql.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16))))
+ON CONFLICT(host, name) DO UPDATE SET
+ spec=excluded.spec, path=excluded.path,
+ next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
+ due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
+ attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
+ registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END
+WHERE timers.kind=excluded.kind`,
+ t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
+ if err != nil {
return err
- case err != nil:
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
return err
}
- if existing.Kind != t.Kind {
+ if n == 0 {
return ErrTimerKind
}
- next, due := existing.NextAt, existing.DueAt
- if existing.Spec != t.Spec || existing.Path != t.Path || t.NextAt != 0 {
- next, due = t.NextAt, t.DueAt
- }
- _, err = r.sql.Exec(`UPDATE timers SET spec=?, path=?, next_at=?, due_at=? WHERE host=? AND name=?`, t.Spec, t.Path, next, due, t.Host, t.Name)
- return err
+ return nil
}
func (r *Registry) GetTimer(host, name string) (*Timer, error) {
@@ -81,6 +83,11 @@ func (r *Registry) DeleteTimer(host, name string) error {
return err
}
+func (r *Registry) DeleteTimerRegistration(host, name, registration string) error {
+ _, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=? AND registration=?`, host, name, registration)
+ return err
+}
+
func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
rows, err := r.sql.Query(q, args...)
if err != nil {
@@ -130,7 +137,7 @@ func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Tim
// delivered (a `carlos schedule run` is its own occurrence, not an early
// delivery of the next natural one).
func (r *Registry) FireTimerNow(host, name string, now int64) error {
- _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=? WHERE host=? AND name=?`, now, now, host, name)
+ _, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=lower(hex(randomblob(16))) WHERE host=? AND name=?`, now, now, host, name)
return err
}
@@ -138,10 +145,14 @@ func (r *Registry) FireTimerNow(host, name string, now int64) error {
// due is the occurrence THAT next delivery is for: the same value as next
// when the row advances, and the row's current DueAt when next is a retry
// of the occurrence just attempted — the runner decides which.
-func (r *Registry) MarkTimerFired(host, name string, at int64, status string, ms int64, attempts int, next, due int64) error {
- _, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=?`,
- at, status, ms, attempts, next, due, host, name)
- return err
+func (r *Registry) MarkTimerFired(host, name, registration string, at int64, status string, ms int64, attempts int, next, due int64) (bool, error) {
+ res, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=? AND registration=?`,
+ at, status, ms, attempts, next, due, host, name, registration)
+ if err != nil {
+ return false, err
+ }
+ n, err := res.RowsAffected()
+ return n != 0, err
}
// SetTimerStatus writes last_status alone — converge's one word on a row
diff --git a/internal/registry/timers_test.go b/internal/registry/timers_test.go
index 57b2eb85c..004cada04 100644
--- a/internal/registry/timers_test.go
+++ b/internal/registry/timers_test.go
@@ -24,7 +24,7 @@ func TestTimersUpsertPreservesRunnerFields(t *testing.T) {
if err := r.UpsertTimer(Timer{Host: "h", Name: "sync", Kind: "schedule", Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: 100, CreatedAt: 1}); err != nil {
t.Fatal(err)
}
- if err := r.MarkTimerFired("h", "sync", 100, "204", 300, 0, 21700, 21700); err != nil {
+ if _, err := r.MarkTimerFired("h", "sync", timerRegistration(t, r, "h", "sync"), 100, "204", 300, 0, 21700, 21700); err != nil {
t.Fatal(err)
}
// Converge sees the same declaration again: spec/path unchanged, next_at
@@ -135,7 +135,7 @@ func TestTimersDueAtFollowsTheOccurrence(t *testing.T) {
t.Fatal(err)
}
// A retry: next moves to the retry instant, due stays the occurrence.
- if err := r.MarkTimerFired("h", "s", 3600, "503", 1, 1, 3660, 3600); err != nil {
+ if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3600, "503", 1, 1, 3660, 3600); err != nil {
t.Fatal(err)
}
got, _ := r.GetTimer("h", "s")
@@ -143,7 +143,7 @@ func TestTimersDueAtFollowsTheOccurrence(t *testing.T) {
t.Fatalf("after retry: %+v", got)
}
// An advance: both move together.
- if err := r.MarkTimerFired("h", "s", 3660, "204", 1, 0, 7200, 7200); err != nil {
+ if _, err := r.MarkTimerFired("h", "s", timerRegistration(t, r, "h", "s"), 3660, "204", 1, 0, 7200, 7200); err != nil {
t.Fatal(err)
}
if got, _ = r.GetTimer("h", "s"); got.NextAt != 7200 || got.DueAt != 7200 {
@@ -288,3 +288,169 @@ func TestEarliestTimersEmpty(t *testing.T) {
t.Errorf("EarliestTimers on an empty table = %v, want empty", got)
}
}
+
+func timerRegistration(t *testing.T, r *Registry, host, name string) string {
+ t.Helper()
+ row, err := r.GetTimer(host, name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return row.Registration
+}
+
+func TestTimerRegistrationProtectsMutations(t *testing.T) {
+ for _, mutation := range []string{"identical", "spec", "path", "manual", "recreate"} {
+ t.Run(mutation, func(t *testing.T) {
+ r := openTimersReg(t)
+ tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100, Attempts: 4, LastAt: 80, LastStatus: "503", LastMs: 1}
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ switch mutation {
+ case "spec":
+ tm.Spec, tm.NextAt, tm.DueAt = "every 2h0m0s", 0, 0
+ case "path":
+ tm.Path, tm.NextAt, tm.DueAt = "/jobs/other", 0, 0
+ case "recreate":
+ if err := r.DeleteTimer(tm.Host, tm.Name); err != nil {
+ t.Fatal(err)
+ }
+ tm.Attempts = 0
+ }
+ if mutation == "manual" {
+ err = r.FireTimerNow(tm.Host, tm.Name, 100)
+ } else {
+ err = r.UpsertTimer(tm)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ replacement, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if replacement.Registration == "" || replacement.Registration == before.Registration || replacement.Attempts != 0 {
+ t.Fatalf("new registration: %+v", replacement)
+ }
+ if _, err := r.MarkTimerFired(tm.Host, tm.Name, before.Registration, 100, "204", 3, 0, 3700, 3700); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.DeleteTimerRegistration(tm.Host, tm.Name, before.Registration); err != nil {
+ t.Fatal(err)
+ }
+ after, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *after != *replacement {
+ t.Fatalf("stale mutation changed registration: %+v -> %+v", replacement, after)
+ }
+ if err := r.DeleteTimerRegistration(tm.Host, tm.Name, after.Registration); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.GetTimer(tm.Host, tm.Name); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("current deletion: %v", err)
+ }
+ })
+ }
+}
+
+func TestTimerRegistrationLegacyMigration(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`ALTER TABLE timers DROP COLUMN registration`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`INSERT INTO timers (host,name,kind,spec,path,next_at,created_at) VALUES ('h','s','once','at 100','/jobs/sync',100,1)`); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < 2; i++ {
+ r, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ row, err := r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration != "" {
+ t.Fatalf("legacy registration: %q", row.Registration)
+ }
+ if _, err := r.MarkTimerFired("h", "s", row.Registration, 100, "503", 3, 1, 160, 100); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ }
+ r, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ row, err := r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 160 || row.DueAt != 100 || row.Attempts != 1 {
+ t.Fatalf("legacy completion: %+v", row)
+ }
+ if err := r.FireTimerNow("h", "s", 100); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.MarkTimerFired("h", "s", "", 100, "204", 3, 0, 0, 0); err != nil {
+ t.Fatal(err)
+ }
+ row, err = r.GetTimer("h", "s")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration == "" || row.NextAt != 100 || row.Attempts != 0 {
+ t.Fatalf("rearmed legacy: %+v", row)
+ }
+}
+
+func TestUnchangedTimerUpsertConcurrentWithCompletion(t *testing.T) {
+ r := openTimersReg(t)
+ tm := Timer{Host: "h", Name: "s", Kind: "schedule", Spec: "every 1h0m0s", Path: "/jobs/sync", NextAt: 100, DueAt: 100}
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ registration := timerRegistration(t, r, tm.Host, tm.Name)
+ tm.NextAt, tm.DueAt = 0, 0
+ for i := int64(1); i <= 100; i++ {
+ start := make(chan struct{})
+ errs := make(chan error, 9)
+ for n := 0; n < 8; n++ {
+ go func() { <-start; errs <- r.UpsertTimer(tm) }()
+ }
+ go func() {
+ <-start
+ _, err := r.MarkTimerFired(tm.Host, tm.Name, registration, i, "503", 1, 1, 100+i, 100)
+ errs <- err
+ }()
+ close(start)
+ for n := 0; n < 9; n++ {
+ if err := <-errs; err != nil {
+ t.Fatal(err)
+ }
+ }
+ row, err := r.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 100+i || row.DueAt != 100 || row.LastAt != i || row.Attempts != 1 || row.Registration != registration {
+ t.Fatalf("completion overwritten at iteration %d: %+v", i, row)
+ }
+ }
+}
diff --git a/internal/relay/authority.go b/internal/relay/authority.go
index f827bce06..beae93c24 100644
--- a/internal/relay/authority.go
+++ b/internal/relay/authority.go
@@ -80,6 +80,11 @@ type Heartbeat struct {
ReadAt time.Time `json:"read_at"`
Listeners []string `json:"listeners"`
Kids []string `json:"kids"`
+ // Retiring is kid -> retire_at for the loaded masters that carry one,
+ // so `carlos relay status` can tell an operator mid-rotation when the
+ // old key may leave the box's masters file (discussion #120). Additive:
+ // a relay older than this field simply reports none.
+ Retiring map[string]time.Time `json:"retiring,omitempty"`
// The two "why is this record stale" fields: the last masters-file
// reload error and the last authority read error, either of which
// explains a ReadAt that has stopped moving.
diff --git a/internal/relay/masters.go b/internal/relay/masters.go
index c8c1f7431..12886dbd8 100644
--- a/internal/relay/masters.go
+++ b/internal/relay/masters.go
@@ -67,6 +67,25 @@ func (m *Masters) Lookup(kid string, now time.Time) (Master, bool) {
return mk, true
}
+// Retiring is the kids that carry a retire_at, and when it falls. An
+// operator mid-rotation reads it through `carlos relay status` to learn when
+// the old key may be dropped from /etc/carlos/relay-masters — the one fact
+// the CLI could not answer before (discussion #120). Kids with no retirement
+// date are absent rather than zero-valued, so "nothing is retiring" is an
+// empty map and not a map of zero times.
+func (m *Masters) Retiring() map[string]time.Time {
+ if m == nil {
+ return nil
+ }
+ out := map[string]time.Time{}
+ for k, mk := range m.byKid {
+ if !mk.RetireAt.IsZero() {
+ out[k] = mk.RetireAt
+ }
+ }
+ return out
+}
+
func (m *Masters) Kids() []string {
out := make([]string, 0, len(m.byKid))
for k := range m.byKid {
diff --git a/internal/relay/masters_test.go b/internal/relay/masters_test.go
index 1213e2fe5..2de1d203b 100644
--- a/internal/relay/masters_test.go
+++ b/internal/relay/masters_test.go
@@ -100,3 +100,26 @@ func TestWatchMastersReloadsOnChange(t *testing.T) {
t.Fatal("no reload within three poll ticks")
}
}
+
+// Retiring is what `carlos relay status` turns into "when may this key leave
+// the box" (discussion #120). A master with no retirement date must be absent
+// rather than present with a zero time, or the CLI prints a retirement for
+// every key a box holds.
+func TestMastersRetiring(t *testing.T) {
+ m, err := ParseMasters(strings.NewReader(
+ "k0 " + base64.StdEncoding.EncodeToString(testMaster()) + " -\n" +
+ "k1 " + base64.StdEncoding.EncodeToString(testMaster()) + " 2026-09-09T12:00:00Z\n"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := m.Retiring()
+ if len(r) != 1 {
+ t.Fatalf("Retiring() = %v, want just k1", r)
+ }
+ if got, want := r["k1"].UTC().Format(time.RFC3339), "2026-09-09T12:00:00Z"; got != want {
+ t.Errorf("k1 retire_at = %s, want %s", got, want)
+ }
+ if _, ok := r["k0"]; ok {
+ t.Error("k0 has no retirement date and must not appear")
+ }
+}
diff --git a/internal/relay/race_off_test.go b/internal/relay/race_off_test.go
new file mode 100644
index 000000000..472c943f7
--- /dev/null
+++ b/internal/relay/race_off_test.go
@@ -0,0 +1,6 @@
+//go:build !race
+
+package relay
+
+// raceDetector is false in an ordinary build. See race_on_test.go.
+const raceDetector = false
diff --git a/internal/relay/race_on_test.go b/internal/relay/race_on_test.go
new file mode 100644
index 000000000..5ce1e39f7
--- /dev/null
+++ b/internal/relay/race_on_test.go
@@ -0,0 +1,12 @@
+//go:build race
+
+package relay
+
+// raceDetector lets one test skip itself under -race. The race it would
+// report is upstream, in pion/turn: Allocation.AddPermission publishes a
+// permission under permissionsLock and then calls perms.start(), which
+// writes p.lifetimeTimer with no lock, while Allocation.Close reads that
+// field from the packet-handler goroutine. Nothing in this package can fix
+// it, and leaving it unskipped costs the whole package its -race coverage,
+// which is worth more than this one test (discussion #120).
+const raceDetector = true
diff --git a/internal/relay/serve.go b/internal/relay/serve.go
index f855f51a5..9c3aaae33 100644
--- a/internal/relay/serve.go
+++ b/internal/relay/serve.go
@@ -124,6 +124,13 @@ type Config struct {
// test seam and nothing else: production is :3478 on both, per spec §3,
// and no env may move them. Empty means the constants above.
udp, tcp string
+
+ // extraAllow is the same kind of seam, for the same reason. A relay on a
+ // real box never permits a loopback peer, and an end-to-end test that
+ // carries traffic between two sockets on this machine has nowhere else
+ // to put them. Unexported so no env and no operator can widen the peer
+ // policy; ServerConfig.ExtraAllow is the reviewed, production-facing one.
+ extraAllow []*net.IPNet
}
// ConfigFromEnv builds a Config from the unit's environment. imdsIP and
@@ -242,20 +249,21 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
tcp = relayTCPAddr
}
srv, err := NewServer(ServerConfig{
- Region: cfg.Region,
- RelayIP: cfg.PublicIP,
- UDP: udp,
- TCP: tcp,
- MinPort: 49152,
- MaxPort: 65535,
- DenyCIDRs: cfg.DenyCIDRs,
- Limits: cfg.Limits,
- Now: now,
- Log: logf,
- Meter: meter,
- Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
- Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
- Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
+ Region: cfg.Region,
+ RelayIP: cfg.PublicIP,
+ UDP: udp,
+ TCP: tcp,
+ MinPort: 49152,
+ MaxPort: 65535,
+ DenyCIDRs: cfg.DenyCIDRs,
+ ExtraAllow: cfg.extraAllow,
+ Limits: cfg.Limits,
+ Now: now,
+ Log: logf,
+ Meter: meter,
+ Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
+ Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
+ Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
})
if err != nil {
return err
@@ -309,6 +317,7 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
ReadAt: lease.Last(), // the last SUCCESSFUL read, never the wall clock
Listeners: listeners,
Kids: kidsOf(masters),
+ Retiring: retiringOf(masters),
MastersError: mastersE,
}
if err != nil {
@@ -437,6 +446,18 @@ func kidsOf(m *Masters) []string {
return m.Kids()
}
+// retiringOf is Masters.Retiring with the same nil guard kidsOf carries.
+func retiringOf(m *Masters) map[string]time.Time {
+ if m == nil {
+ return nil
+ }
+ r := m.Retiring()
+ if len(r) == 0 {
+ return nil
+ }
+ return r
+}
+
// readPSI reads the `some avg10` percentage out of a pressure-stall file
// (/proc/pressure/memory):
//
diff --git a/internal/relay/serve_test.go b/internal/relay/serve_test.go
index 565b42328..6008e1ed1 100644
--- a/internal/relay/serve_test.go
+++ b/internal/relay/serve_test.go
@@ -7,6 +7,7 @@ import (
"net"
"os"
"path/filepath"
+ "strconv"
"strings"
"sync"
"testing"
@@ -188,7 +189,11 @@ func TestServeWritesHeartbeatAndStops(t *testing.T) {
store := newMemStore(t)
dir := t.TempDir()
masters := filepath.Join(dir, "relay-masters")
- if err := os.WriteFile(masters, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil {
+ // k2 carries a retirement date, so this fixture also covers the hop that
+ // puts it on the heartbeat for `carlos relay status` (discussion #120).
+ if err := os.WriteFile(masters, []byte(
+ "k1 "+b64(testMaster())+" -\n"+
+ "k2 "+b64(testMaster())+" 2026-09-09T12:00:00Z\n"), 0o640); err != nil {
t.Fatal(err)
}
cfg := Config{
@@ -227,6 +232,28 @@ func TestServeWritesHeartbeatAndStops(t *testing.T) {
if hb.ReadAt.IsZero() {
t.Error("heartbeat carries no read time, so the lease clock is invisible")
}
+ // Without this the relay half of the retiring display can be deleted with
+ // the whole suite green, and an operator mid-rotation silently reads
+ // "nothing is retiring". Polled rather than asserted on the heartbeat
+ // above: the masters watcher is a goroutine, so the FIRST heartbeat can
+ // legitimately be written before the file has been read.
+ deadline := time.Now().Add(10 * time.Second)
+ for {
+ hb := waitHeartbeat(t, store, "box-serve")
+ if got, ok := hb.Retiring["k2"]; ok {
+ if want := "2026-09-09T12:00:00Z"; got.UTC().Format(time.RFC3339) != want {
+ t.Errorf("k2 retire_at = %s, want %s", got.UTC().Format(time.RFC3339), want)
+ }
+ if _, ok := hb.Retiring["k1"]; ok {
+ t.Error("k1 has no retirement date and must not be reported as retiring")
+ }
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("no heartbeat ever carried k2's retirement (last: %+v)", hb.Retiring)
+ }
+ time.Sleep(50 * time.Millisecond)
+ }
cancel()
select {
@@ -425,3 +452,167 @@ func waitHeartbeat(t *testing.T, store release.Store, box string) Heartbeat {
time.Sleep(20 * time.Millisecond)
}
}
+
+// heartbeatUDPPort pulls the udp listener's port out of a heartbeat, which is
+// the only place a test outside Serve can learn the port it bound.
+func heartbeatUDPPort(t *testing.T, hb Heartbeat) int {
+ t.Helper()
+ for _, l := range hb.Listeners {
+ if p, ok := strings.CutPrefix(l, "udp:"); ok {
+ n, err := strconv.Atoi(p)
+ if err != nil {
+ t.Fatalf("udp listener %q has no port: %v", l, err)
+ }
+ return n
+ }
+ }
+ t.Fatalf("heartbeat lists no udp listener: %v", hb.Listeners)
+ return 0
+}
+
+// TestServeEndsAllocationsOnDisable is the end-to-end disable path through
+// the real Serve loop, which spec §7 asks for and the branch never carried
+// (discussion #120): a pion client holding a live allocation, an app removed
+// from the authority record, and the relay acting on it within one authority
+// read rather than on a direct EndWhere call.
+//
+// The serial is what makes it a test and not a minute-long wait: the relay
+// polls control/serial every three seconds and re-reads the authority record
+// when it moves, which is the same path a console-mediated disable takes on
+// a real box.
+func TestServeEndsAllocationsOnDisable(t *testing.T) {
+ store := newMemStore(t)
+ dir := t.TempDir()
+ mastersPath := filepath.Join(dir, "relay-masters")
+ if err := os.WriteFile(mastersPath, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil {
+ t.Fatal(err)
+ }
+ ctx := context.Background()
+ putAuthority := func(apps []AuthorityApp) time.Time {
+ t.Helper()
+ writtenAt := time.Now().UTC()
+ body, err := json.Marshal(Authority{Apps: apps, WrittenAt: writtenAt})
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, etag, _ := store.Get(ctx, AuthorityKey)
+ if _, err := store.Put(ctx, AuthorityKey, body, etag); err != nil {
+ t.Fatal(err)
+ }
+ if err := release.BumpSerial(ctx, store, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ return writtenAt
+ }
+ putAuthority([]AuthorityApp{{Account: "abc", App: "meet", Incarnation: "0123456789abcdef", Regions: []string{"eu-west-1"}}})
+
+ cfg := Config{
+ Box: "box-disable", Region: "eu-west-1", MastersPath: mastersPath,
+ // The advertised relay address is deliberately NOT loopback, as on a
+ // real box: the peers are on loopback and a relay IP equal to them
+ // does not forward (TestServerRelaysBothLegs makes the same choice).
+ PublicIP: net.ParseIP("203.0.113.7"), Store: store, Limits: DefaultLimits(),
+ PressurePath: filepath.Join(dir, "no-pressure-here"),
+ udp: "127.0.0.1:0",
+ tcp: "127.0.0.1:0",
+ extraAllow: mustCIDRs("127.0.0.0/8"),
+ }
+ var logMu sync.Mutex
+ quiet := false
+ logf := func(format string, args ...any) {
+ logMu.Lock()
+ defer logMu.Unlock()
+ if !quiet {
+ t.Logf(format, args...)
+ }
+ }
+ t.Cleanup(func() { logMu.Lock(); quiet = true; logMu.Unlock() })
+
+ runCtx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- Serve(runCtx, cfg, logf) }()
+
+ port := heartbeatUDPPort(t, waitHeartbeat(t, store, "box-disable"))
+ username, password := creds(time.Now(), "abc", "meet", "0123456789abcdef")
+ client := turnClient(t, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, username, password)
+ relayConn, err := client.Allocate()
+ if err != nil {
+ t.Fatalf("allocate against a relay that has read an enabling record: %v", err)
+ }
+ defer relayConn.Close()
+
+ // A peer on loopback, permitted through the extraAllow seam, and one
+ // packet through the relay to prove the allocation carries traffic while
+ // the app IS enabled. Without this the "no traffic after" check below
+ // would pass against a relay that never forwarded anything at all.
+ peer, err := net.ListenPacket("udp4", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer peer.Close()
+ peerAddr := peer.LocalAddr().(*net.UDPAddr)
+ if err := client.CreatePermission(peerAddr); err != nil {
+ t.Fatalf("permission for a loopback peer: %v", err)
+ }
+ if _, err := relayConn.WriteTo([]byte("before"), peerAddr); err != nil {
+ t.Fatalf("write through the relay: %v", err)
+ }
+ buf := make([]byte, 64)
+ peer.SetReadDeadline(time.Now().Add(5 * time.Second))
+ if n, _, err := peer.ReadFrom(buf); err != nil || string(buf[:n]) != "before" {
+ t.Fatalf("the relay did not carry traffic while enabled: n=%d err=%v", n, err)
+ }
+
+ // The disable: the same record without the app, and a serial bump so the
+ // relay re-reads inside three seconds rather than at the next minute.
+ disabledAt := putAuthority(nil)
+
+ // Wait on the relay's OWN report that it is holding the new record —
+ // the heartbeat carries the WrittenAt of the authority copy it serves —
+ // rather than on an allocation failing. Retrying Allocate until any
+ // error appeared was the first shape of this test and it was not a test
+ // at all: a refusal for an unrelated reason (a per-app allocation cap,
+ // say) reads exactly like the disable, so the assertion passed with the
+ // app still enabled.
+ deadline := time.Now().Add(20 * time.Second)
+ for {
+ hb := waitHeartbeat(t, store, "box-disable")
+ if !hb.WrittenAt.Before(disabledAt) {
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("the relay never picked up the record written at %s (last heartbeat holds %s)", disabledAt, hb.WrittenAt)
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+
+ // The allocation taken before the disable must stop carrying traffic.
+ // This is the assertion the test is named for, and the first version of
+ // it did not make: it only re-Allocated, which proves admission refuses
+ // a NEW allocation and says nothing about the live one. Deleting Serve's
+ // EndWhere(..., "authority") call left that version green.
+ if _, err := relayConn.WriteTo([]byte("after"), peerAddr); err == nil {
+ peer.SetReadDeadline(time.Now().Add(time.Second))
+ if n, _, err := peer.ReadFrom(buf); err == nil {
+ t.Errorf("the relay forwarded %q after the app was disabled: the allocation did not end", buf[:n])
+ }
+ }
+
+ // And a credential that was good a moment ago must be refused, which is
+ // what a member sees after `relay disable`.
+ probe := turnClient(t, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port}, username, password)
+ if _, err := probe.Allocate(); err == nil {
+ t.Error("the relay allocated for an app the authority record no longer carries")
+ }
+
+ cancel()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("Serve returned %v, want nil on a cancelled context", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("Serve did not return after its context was cancelled")
+ }
+}
diff --git a/internal/relay/server_test.go b/internal/relay/server_test.go
index 043d53bee..0b69675e4 100644
--- a/internal/relay/server_test.go
+++ b/internal/relay/server_test.go
@@ -657,3 +657,40 @@ func TestServerCloseKeepsReplacementBinding(t *testing.T) {
t.Fatal("the closed socket is still owned, so its admission never came back")
}
}
+
+// TestServerRefusesPermissionIntoConfiguredDenyCIDR covers the operator's
+// half of the peer policy, which the built-in private ranges do not reach:
+// CARLOS_RELAY_DENY_CIDRS carries the box's own VPC CIDR, and that CIDR is
+// ordinary public-looking space as far as the built-in list is concerned.
+// It is the one thing standing between a TURN allocation and the deployment's
+// private network, and until now only the built-in ranges had a test
+// (discussion #120).
+func TestServerRefusesPermissionIntoConfiguredDenyCIDR(t *testing.T) {
+ if raceDetector {
+ t.Skip("pion/turn races on Permission.lifetimeTimer between AddPermission and Close; see race_on_test.go")
+ }
+ // 198.51.100.0/24 stands in for the VPC: globally routable as far as
+ // the built-in private list is concerned, denied only because the
+ // operator said so.
+ _, vpc, err := net.ParseCIDR("198.51.100.0/24")
+ if err != nil {
+ t.Fatal(err)
+ }
+ s, _, now, _, _ := newTestServer(t, "127.0.0.1", nil, func(cfg *ServerConfig) {
+ cfg.DenyCIDRs = append(cfg.DenyCIDRs, vpc)
+ })
+ username, password := creds(now(), "abc", "meet", "0123456789abcdef")
+ client := turnClient(t, s.UDPAddr(), username, password)
+ if _, err := client.Allocate(); err != nil {
+ t.Fatalf("allocate: %v", err)
+ }
+ if err := client.CreatePermission(&net.UDPAddr{IP: net.ParseIP("198.51.100.9"), Port: 1}); err == nil {
+ t.Error("permission into the configured deny CIDR was granted — a TURN client could reach the VPC")
+ }
+ // The control: an address outside both the built-in list and the
+ // configured range is still permitted, so the test is not passing
+ // because permissions are broken outright.
+ if err := client.CreatePermission(&net.UDPAddr{IP: net.ParseIP("203.0.113.9"), Port: 1}); err != nil {
+ t.Errorf("permission to an ordinary public peer was refused: %v", err)
+ }
+}
diff --git a/internal/schedule/control.go b/internal/schedule/control.go
index 14e663f5f..acd30bd2f 100644
--- a/internal/schedule/control.go
+++ b/internal/schedule/control.go
@@ -56,6 +56,13 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}
}
+ mux.HandleFunc("GET /schedule-capabilities", authed(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ json.NewEncoder(w).Encode(struct {
+ RegistrationFence bool `json:"registration_fence"`
+ }{RegistrationFence: true})
+ }))
+
mux.HandleFunc("PUT /schedule/{name}", authed(func(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
if !ValidName(name) {
@@ -112,7 +119,11 @@ func Mount(mux *http.ServeMux, host string, store TimerStore, token func() (stri
}
}
sp := Spec{Kind: KindOnce, At: atT.UTC()}
- t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: atT.Unix(), DueAt: atT.Unix(), CreatedAt: now().Unix()}
+ next := atT.Unix()
+ if next <= 0 {
+ next = max(int64(1), now().Unix())
+ }
+ t := registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: sp.String(), Path: body.Path, NextAt: next, DueAt: atT.Unix(), CreatedAt: now().Unix()}
if err := store.UpsertTimer(t); err != nil {
http.Error(w, "registry error", http.StatusInternalServerError)
return
diff --git a/internal/schedule/control_test.go b/internal/schedule/control_test.go
index 23909e324..d58667c2b 100644
--- a/internal/schedule/control_test.go
+++ b/internal/schedule/control_test.go
@@ -193,3 +193,66 @@ func TestControlCapsOncePerHost(t *testing.T) {
}
func itoa(n int64) string { return strconv.FormatInt(n, 10) }
+
+func TestControlScheduleCapabilities(t *testing.T) {
+ _, mux := controlFixture(t)
+ for _, auth := range []string{"", "Bearer wrong", "Basic tok"} {
+ rec := do(mux, "GET", "/schedule-capabilities", auth, "")
+ if rec.Code != http.StatusUnauthorized || strings.Contains(rec.Body.String(), "registration_fence") {
+ t.Fatalf("unauthorized capability response: %d %s", rec.Code, rec.Body.String())
+ }
+ }
+ rec := do(mux, "GET", "/schedule-capabilities", "Bearer tok", "")
+ if rec.Code != http.StatusOK || rec.Header().Get("Content-Type") != "application/json" {
+ t.Fatalf("capability response: %d %s", rec.Code, rec.Body.String())
+ }
+ var got map[string]bool
+ if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 1 || !got["registration_fence"] {
+ t.Fatalf("capabilities: %v", got)
+ }
+ old := http.NewServeMux()
+ old.HandleFunc("GET /schedule", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
+ rec = do(old, "GET", "/schedule-capabilities", "Bearer tok", "")
+ if rec.Code != http.StatusNotFound || strings.Contains(rec.Body.String(), "registration_fence") {
+ t.Fatalf("old capability response: %d %s", rec.Code, rec.Body.String())
+ }
+}
+
+func TestControlNonpositiveOccurrenceArmsTimer(t *testing.T) {
+ for _, requested := range []int64{0, -1, -62135596800} {
+ for _, current := range []int64{1787490420, 0, -1} {
+ t.Run(strconv.FormatInt(requested, 10)+"/"+strconv.FormatInt(current, 10), func(t *testing.T) {
+ reg, _ := controlFixture(t)
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, func() time.Time { return time.Unix(current, 0) })
+ body := `{"at":"` + time.Unix(requested, 0).UTC().Format(time.RFC3339) + `","path":"/jobs/remind"}`
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", body)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != max(int64(1), current) || row.DueAt != requested || row.Spec != "at "+strconv.FormatInt(requested, 10) {
+ t.Fatalf("normalized timer: %+v", row)
+ }
+ first := row.Registration
+ rec = do(mux, "PUT", "/schedule/remind", "Bearer tok", body)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("repeat PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Registration == first {
+ t.Fatal("repeat PUT did not rearm")
+ }
+ })
+ }
+ }
+}
diff --git a/internal/schedule/converge_test.go b/internal/schedule/converge_test.go
index c8375bf39..95603d9e4 100644
--- a/internal/schedule/converge_test.go
+++ b/internal/schedule/converge_test.go
@@ -131,7 +131,7 @@ func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
// A second pass with nothing changed: no new reports, next_at untouched.
reports = nil
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 10, 0, 999, 999)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 10, 0, 999, 999)
if err := c.Pass(ctx); err != nil {
t.Fatal(err)
}
@@ -190,7 +190,7 @@ func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
t.Fatalf("%s not fired: %+v", h, tm)
}
}
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
c2.Pass(ctx) // same etag: must not re-fire
if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
t.Fatal("run signal replayed on an unchanged etag")
@@ -511,7 +511,7 @@ func TestConvergeRunSignalGoingAbsentIsNotAFire(t *testing.T) {
if err := c.Pass(ctx); err != nil {
t.Fatal(err)
}
- reg.MarkTimerFired("a.bab.oncarlos.com", "sync", now.Unix(), "204", 1, 0, 777, 777)
+ reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
t.Fatalf("setup: run signal did not fire: %+v", tm)
}
@@ -597,3 +597,12 @@ func TestConvergeParksAScheduleThatNeverFires(t *testing.T) {
t.Fatalf("a parked row is due: %+v", due)
}
}
+
+func scheduleTimerRegistration(t *testing.T, reg *registry.Registry, host, name string) string {
+ t.Helper()
+ row, err := reg.GetTimer(host, name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return row.Registration
+}
diff --git a/internal/schedule/runner.go b/internal/schedule/runner.go
index 888a5a784..cfc6dfd46 100644
--- a/internal/schedule/runner.go
+++ b/internal/schedule/runner.go
@@ -308,11 +308,20 @@ func (r *Runner) slots() chan struct{} {
// sleeping. Nothing in the agent calls it.
func (r *Runner) wait() { r.inflight.Wait() }
+func (r *Runner) timerStillDue(t registry.Timer) bool {
+ current, err := r.Reg.GetTimer(t.Host, t.Name)
+ return err == nil && current.Registration == t.Registration && current.NextAt == t.NextAt && current.NextAt > 0 && current.NextAt <= r.now().Unix()
+}
+
func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
+ if !r.timerStillDue(t) {
+ return
+ }
+
sp, err := ParseSpec(t.Spec)
if err != nil {
log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err)
- r.Reg.DeleteTimer(rt.Host, t.Name)
+ r.Reg.DeleteTimerRegistration(rt.Host, t.Name, t.Registration)
return
}
start := r.now()
@@ -328,6 +337,9 @@ func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer)
r.record(rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0)
return
}
+ if !r.timerStillDue(t) {
+ return
+ }
dctx, cancel := context.WithTimeout(ctx, r.timeout())
req = req.WithContext(dctx)
status, derr := r.Deliver.Deliver(dctx, rt, req)
@@ -379,12 +391,13 @@ func (r *Runner) record(rt registry.Route, t registry.Timer, sp Spec, at time.Ti
} else if !isSuccess(o.Status) {
level = "warn"
}
- if err := r.Reg.MarkTimerFired(rt.Host, t.Name, at.Unix(), status, took.Milliseconds(), attempts, next, due); err != nil {
+ applied, err := r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due)
+ if err != nil {
log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
}
if r.Event != nil {
msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))
- if attempts > 0 {
+ if attempts > 0 && applied {
msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339))
}
r.Event(rt, level, msg)
@@ -420,13 +433,15 @@ func classify(status int, err error, timeout time.Duration) Outcome {
// Classify is the exported policy for the host package's tests.
func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) }
-// dueOf is the occurrence a row is delivering: DueAt, or NextAt for a row
-// written before due_at existed (whose NextAt, at attempt 0, IS the
-// occurrence — and once pinned by a retry it stays the occurrence).
func dueOf(t registry.Timer) int64 {
if t.DueAt != 0 {
return t.DueAt
}
+ if t.Kind == KindOnce {
+ if sp, err := ParseSpec(t.Spec); err == nil && sp.Kind == KindOnce && sp.At.Unix() == 0 {
+ return 0
+ }
+ }
return t.NextAt
}
diff --git a/internal/schedule/runner_test.go b/internal/schedule/runner_test.go
index a6880a9f9..3602c996f 100644
--- a/internal/schedule/runner_test.go
+++ b/internal/schedule/runner_test.go
@@ -5,6 +5,8 @@ import (
"errors"
"fmt"
"net/http"
+ "net/http/httptest"
+ "os"
"path/filepath"
"strings"
"sync"
@@ -349,7 +351,8 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
reg, r := runnerFixture(t, d)
now := at("2026-08-23T13:07:00Z")
- r.Now = func() time.Time { return now }
+ clock := &testClock{t: now}
+ r.Now = clock.now
hosts := []string{"slow.bab.oncarlos.com", "b.bab.oncarlos.com", "c.bab.oncarlos.com", "d.bab.oncarlos.com", "e.bab.oncarlos.com", "f.bab.oncarlos.com", "g.bab.oncarlos.com"}
for i, h := range hosts {
if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
@@ -395,7 +398,7 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
}
// The parked host's row is still due; a second tick must skip it.
- now = now.Add(time.Minute)
+ clock.advance(time.Minute)
r.Tick(context.Background())
close(d.gate)
r.wait()
@@ -406,7 +409,7 @@ func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
t.Fatalf("slow row after release: %+v", tm)
}
// And once released, it is fair game again.
- now = at("2026-08-23T14:00:01Z")
+ clock.advance(at("2026-08-23T14:00:01Z").Sub(clock.now()))
r.tick()
if n := d.firedHosts()[d.slow]; n != 2 {
t.Fatalf("released host delivered %d times, want 2", n)
@@ -672,3 +675,304 @@ func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {
t.Fatalf("delivered to %d hosts, want %d", len(fired), hosts)
}
}
+
+type timerMutationDeliverer struct {
+ mutate func()
+ status int
+}
+
+func (d timerMutationDeliverer) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
+ d.mutate()
+ return d.status, nil
+}
+
+func TestDeliveryCompletionPreservesReplacementRegistration(t *testing.T) {
+ for _, mutation := range []string{"identical", "timing", "path", "cancel", "recreate", "manual"} {
+ for _, status := range []int{204, 503} {
+ t.Run(fmt.Sprintf("%s/%d", mutation, status), func(t *testing.T) {
+ reg, runner := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ const host = "a.bab.oncarlos.com"
+ mux := http.NewServeMux()
+ Mount(mux, host, reg, func() (string, error) { return "tok", nil }, runner.Now)
+ put := func(when time.Time, path string) {
+ req := httptest.NewRequest(http.MethodPut, "/schedule/remind", strings.NewReader(fmt.Sprintf(`{"at":%q,"path":%q}`, when.Format(time.RFC3339), path)))
+ req.Header.Set("Authorization", "Bearer tok")
+ w := httptest.NewRecorder()
+ mux.ServeHTTP(w, req)
+ if w.Code != http.StatusOK {
+ t.Errorf("PUT: %d %s", w.Code, w.Body.String())
+ }
+ }
+ put(now, "/jobs/remind")
+ before, err := reg.GetTimer(host, "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var replacement *registry.Timer
+ var events []string
+ runner.Event = func(_ registry.Route, _, msg string) { events = append(events, msg) }
+ runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
+ switch mutation {
+ case "identical":
+ put(now, "/jobs/remind")
+ case "timing":
+ put(now.Add(time.Hour), "/jobs/remind")
+ case "path":
+ put(now, "/jobs/changed")
+ case "cancel", "recreate":
+ req := httptest.NewRequest(http.MethodDelete, "/schedule/remind", nil)
+ req.Header.Set("Authorization", "Bearer tok")
+ w := httptest.NewRecorder()
+ mux.ServeHTTP(w, req)
+ if w.Code != http.StatusNoContent {
+ t.Errorf("DELETE: %d", w.Code)
+ }
+ if mutation == "recreate" {
+ put(now, "/jobs/remind")
+ }
+ case "manual":
+ if err := reg.FireTimerNow(host, "remind", now.Unix()); err != nil {
+ t.Error(err)
+ }
+ }
+ replacement, err = reg.GetTimer(host, "remind")
+ if mutation != "cancel" && err != nil {
+ t.Error(err)
+ }
+ }}
+ runner.tick()
+ if len(events) != 1 || strings.Contains(events[0], "retrying at") {
+ t.Fatalf("superseded delivery event: %v", events)
+ }
+ after, err := reg.GetTimer(host, "remind")
+ if mutation == "cancel" {
+ if !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("cancelled timer: %+v, %v", after, err)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if replacement == nil {
+ t.Fatal("replacement missing")
+ }
+ if after.Registration == before.Registration || after.Registration == "" {
+ t.Fatalf("registration did not change: %+v", after)
+ }
+ if *after != *replacement {
+ t.Fatalf("completion changed replacement: got %+v, want %+v", after, replacement)
+ }
+ runner.Deliver = &fakeDeliverer{status: 204}
+ now = time.Unix(after.NextAt, 0)
+ runner.tick()
+ finished, err := reg.GetTimer(host, "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if finished.NextAt != 0 || finished.LastStatus != "204" {
+ t.Fatalf("replacement did not fire: %+v", finished)
+ }
+ })
+ }
+ }
+}
+
+func TestDeliveryCompletionSurvivesUnchangedDeclaration(t *testing.T) {
+ for _, status := range []int{204, 503} {
+ t.Run(fmt.Sprint(status), func(t *testing.T) {
+ reg, runner := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix(), DueAt: now.Unix()}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tm.NextAt, tm.DueAt = 0, 0
+ runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Error(err)
+ }
+ }}
+ runner.tick()
+ after, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if after.Registration != before.Registration || after.LastStatus != fmt.Sprint(status) {
+ t.Fatalf("completion missing: %+v", after)
+ }
+ if status == 503 && (after.Attempts != 1 || after.DueAt != before.DueAt || after.NextAt != now.Add(time.Minute).Unix()) {
+ t.Fatalf("retry changed: %+v", after)
+ }
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ converged, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *converged != *after {
+ t.Fatalf("converge changed completed state: got %+v, want %+v", converged, after)
+ }
+ })
+ }
+}
+
+func TestInvalidTimerSnapshotCannotDeleteReplacement(t *testing.T) {
+ reg, runner := runnerFixture(t, &fakeDeliverer{status: 204})
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "invalid", Path: "/jobs/remind", NextAt: 100}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ old, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ tm.Spec = "at 100"
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ replacement, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ runner.fire(context.Background(), registry.Route{Host: tm.Host}, *old)
+ after, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if *after != *replacement {
+ t.Fatalf("invalid snapshot changed replacement: got %+v, want %+v", after, replacement)
+ }
+}
+
+func TestBatchSkipsTimerChangedByEarlierDelivery(t *testing.T) {
+ for _, mutation := range []string{"cancel", "rearm"} {
+ t.Run(mutation, func(t *testing.T) {
+ reg, runner := runnerFixture(t, nil)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ const host = "a.bab.oncarlos.com"
+ for _, name := range []string{"first", "later"} {
+ if err := reg.UpsertTimer(registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: "at 100", Path: "/jobs/" + name, NextAt: 100}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ calls := 0
+ runner.Deliver = timerMutationDeliverer{status: 204, mutate: func() {
+ calls++
+ if mutation == "cancel" {
+ if err := reg.DeleteTimer(host, "later"); err != nil {
+ t.Error(err)
+ }
+ } else {
+ if err := reg.FireTimerNow(host, "later", now.Add(time.Hour).Unix()); err != nil {
+ t.Error(err)
+ }
+ }
+ }}
+ runner.tick()
+ if calls != 1 {
+ t.Fatalf("delivered %d timers, want first only", calls)
+ }
+ })
+ }
+}
+
+func TestConcurrentTicksSkipSnapshotChangedDuringTokenLookup(t *testing.T) {
+ for _, mutation := range []string{"cancel", "rearm", "retry", "retry-still-due", "done"} {
+ t.Run(mutation, func(t *testing.T) {
+ deliver := &fakeDeliverer{status: 204}
+ reg, runner := runnerFixture(t, deliver)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 100", Path: "/jobs/remind", NextAt: 100}
+ if err := reg.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ before, err := reg.GetTimer(tm.Host, tm.Name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ entered, release := make(chan struct{}), make(chan struct{})
+ var tokenOnce sync.Once
+ runner.Token = func(string) (string, error) { tokenOnce.Do(func() { close(entered) }); <-release; return "tok", nil }
+ runner.Tick(context.Background())
+ <-entered
+ switch mutation {
+ case "cancel":
+ err = reg.DeleteTimer(tm.Host, tm.Name)
+ case "rearm":
+ err = reg.FireTimerNow(tm.Host, tm.Name, now.Add(time.Hour).Unix())
+ case "retry":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Add(time.Minute).Unix(), 100)
+ case "retry-still-due":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Unix()-1, 100)
+ case "done":
+ _, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "204", 0, 0, 0, 0)
+ }
+ if err != nil {
+ close(release)
+ runner.wait()
+ t.Fatal(err)
+ }
+ runner.Tick(context.Background())
+ close(release)
+ runner.wait()
+ runner.fire(context.Background(), registry.Route{Host: tm.Host}, *before)
+ if len(deliver.calls) != 0 {
+ t.Fatalf("delivered stale snapshot %d times", len(deliver.calls))
+ }
+ })
+ }
+}
+
+func TestNonpositiveOccurrenceDeliveredAndRetriedWithOriginalTimestamp(t *testing.T) {
+ for _, requested := range []int64{0, -1, -62135596800} {
+ t.Run(fmt.Sprint(requested), func(t *testing.T) {
+ deliver := &fakeDeliverer{status: 503}
+ reg, runner := runnerFixture(t, deliver)
+ now := at("2026-08-23T13:07:00Z")
+ runner.Now = func() time.Time { return now }
+ mux := http.NewServeMux()
+ Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, runner.Now)
+ rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"`+time.Unix(requested, 0).UTC().Format(time.RFC3339)+`","path":"/jobs/remind"}`)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
+ }
+ runner.tick()
+ row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.Attempts != 1 || row.NextAt != now.Add(time.Minute).Unix() || row.DueAt != requested {
+ t.Fatalf("retry: %+v", row)
+ }
+ now = now.Add(time.Minute)
+ deliver.status = 204
+ runner.tick()
+ if len(deliver.calls) != 2 {
+ t.Fatalf("deliveries: %d", len(deliver.calls))
+ }
+ for _, req := range deliver.calls {
+ if got := req.Header.Get("X-Carlos-Schedule-At"); got != fmt.Sprint(requested) {
+ t.Fatalf("occurrence: %q, want %d", got, requested)
+ }
+ }
+ row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if row.NextAt != 0 || row.LastStatus != "204" {
+ t.Fatalf("completed: %+v", row)
+ }
+ })
+ }
+}
diff --git a/internal/status/snapshot.go b/internal/status/snapshot.go
index 5f1de396d..dd757d7dd 100644
--- a/internal/status/snapshot.go
+++ b/internal/status/snapshot.go
@@ -123,6 +123,13 @@ type Probes struct {
// here means "no aliases known this tick," identical on the wire to "no
// alias rows exist."
Aliases []registry.RouteAlias
+
+ // DBBytes sizes one route's SQLite database on local disk: the file at
+ // registry.Route.DBPath plus its -wal. CollectBox calls it for every
+ // non-platform, non-static route with a DBPath. A nil func, or a nil
+ // answer (no file yet, stat failed), means unknown and the host is left
+ // out of Metrics.RouteDBBytes rather than published as 0.
+ DBBytes func(dbPath string) *int64
}
// UpstreamState is one upstream's health as the edge's prober currently sees
@@ -234,6 +241,13 @@ type BoxMetrics struct {
// present-but-empty map) when this box has no alias rows, same
// "absent, not empty" rule as RouteVersions.
Aliases map[string]AliasReading `json:"aliases,omitempty"`
+ // RouteDBBytes is host -> on-disk size of that route's SQLite database
+ // (file plus -wal), for this box's own routes with a DBPath. It is
+ // RouteVersions' twin: hostname-keyed, so it lives in this private block
+ // and nowhere else. The console's instance and route listings are the
+ // readers. A host whose file could not be sized is absent, never 0; the
+ // whole map is omitted when no route could be sized.
+ RouteDBBytes map[string]int64 `json:"route_db_bytes,omitempty"`
}
// AliasReading is one alias hostname's status as CollectBox saw it this
@@ -458,6 +472,9 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box {
// both fold into the same map — nil (not an empty map) until the first
// entry, same "nothing to report" convention as routeVersions.
var aliasReadings map[string]AliasReading
+ // routeDBBytes is Metrics.RouteDBBytes, built in the same loop and nil
+ // until the first entry, like routeVersions.
+ var routeDBBytes map[string]int64
for _, r := range routes {
// Platform-minted routes (internal/domains) are additional NAMES for
@@ -477,6 +494,14 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box {
}
routeVersions[r.Host] = r.Version
}
+ if r.Kind != "static" && r.DBPath != "" && p.DBBytes != nil {
+ if n := p.DBBytes(r.DBPath); n != nil {
+ if routeDBBytes == nil {
+ routeDBBytes = map[string]int64{}
+ }
+ routeDBBytes[r.Host] = *n
+ }
+ }
// Non-platform routes carry hostname readings too. A first-route
// claim (claims.go: "an empty route means this host is its own
// primary") materializes a ROUTE, not an alias row, so without this
@@ -590,7 +615,8 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box {
if p.MemTotalBytes != nil || p.MemAvailBytes != nil || p.Load1 != nil ||
p.DiskFreeBytes != nil || p.Release != "" || p.Tunnel != "" ||
p.InstanceID != "" || p.Platform != "" || p.SystemChannel != "" || p.System != nil ||
- len(routeVersions) > 0 || len(aliasReadings) > 0 || len(p.Upstreams) > 0 {
+ len(routeVersions) > 0 || len(aliasReadings) > 0 || len(p.Upstreams) > 0 ||
+ len(routeDBBytes) > 0 {
box.Metrics = &BoxMetrics{
MemTotalBytes: p.MemTotalBytes, MemAvailBytes: p.MemAvailBytes,
Load1: p.Load1, DiskFreeBytes: p.DiskFreeBytes,
@@ -604,7 +630,8 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box {
// for a later probe tick to mutate under the marshaller. An empty
// map (rather than nil) reaching this line can only happen
// alongside another known field above, and omitempty drops it.
- Upstreams: p.Upstreams,
+ Upstreams: p.Upstreams,
+ RouteDBBytes: routeDBBytes,
}
}
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
index f85ece127..c16eb93d4 100644
--- a/internal/status/status_test.go
+++ b/internal/status/status_test.go
@@ -1335,3 +1335,58 @@ func TestCollectBoxOmitsPlatformRoutesFromReadings(t *testing.T) {
t.Fatal("the custom domain must appear")
}
}
+
+// TestCollectBoxCarriesRouteDBBytes: every non-platform, non-static route
+// with a DBPath is sized through Probes.DBBytes into Metrics.RouteDBBytes. A
+// route the probe cannot size is absent rather than 0, and the hostname never
+// reaches the public fleet.json.
+func TestCollectBoxCarriesRouteDBBytes(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "sized.test", Kind: "instance", DBPath: "/data/sized.test.db"},
+ {Host: "missing.test", Kind: "instance", DBPath: "/data/missing.test.db"},
+ {Host: "nodb.test", Kind: "instance"},
+ {Host: "static.test", Kind: "static", DBPath: "/data/static.test.db"},
+ {Host: "platform.test", Kind: "instance", DBPath: "/data/sized.test.db", Platform: true},
+ }
+ var asked []string
+ probe := func(p string) *int64 {
+ asked = append(asked, p)
+ if p == "/data/sized.test.db" {
+ n := int64(4096)
+ return &n
+ }
+ return nil
+ }
+ b := status.CollectBox(routes, status.Probes{MachineID: "m", DBBytes: probe}, time.Now())
+ if b.Metrics == nil {
+ t.Fatal("metrics block missing — RouteDBBytes should have attached it")
+ }
+ if len(b.Metrics.RouteDBBytes) != 1 || b.Metrics.RouteDBBytes["sized.test"] != 4096 {
+ t.Fatalf("route_db_bytes = %v, want exactly {sized.test: 4096}", b.Metrics.RouteDBBytes)
+ }
+ if strings.Join(asked, ",") != "/data/sized.test.db,/data/missing.test.db" {
+ t.Errorf("probed %v, want only the two instance routes with a DBPath", asked)
+ }
+
+ s := teststore(t)
+ ctx := context.Background()
+ if err := status.WriteBox(ctx, s, b); err != nil {
+ t.Fatal(err)
+ }
+ f, err := status.BuildFleet(ctx, s, routes, time.Now(), 5*time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fj, err := json.Marshal(f)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(string(fj), "sized.test") {
+ t.Errorf("fleet.json leaked a route_db_bytes hostname: %s", fj)
+ }
+
+ // No probe at all: nothing sized, and no Metrics block from this alone.
+ if b := status.CollectBox(routes[:1], status.Probes{MachineID: "m"}, time.Now()); b.Metrics != nil {
+ t.Errorf("metrics = %+v with no DBBytes probe, want absent", b.Metrics)
+ }
+}
diff --git a/relay.go b/relay.go
index f678ea8ca..b9c8492fb 100644
--- a/relay.go
+++ b/relay.go
@@ -33,6 +33,7 @@ import (
"log"
"os"
"os/signal"
+ "sort"
"strings"
"syscall"
@@ -402,6 +403,23 @@ func printRelayStatus(w io.Writer, res apiRelayReply) {
if len(rg.Kids) > 0 {
line += ", masters " + strings.Join(rg.Kids, ",")
}
+ // A retiring master is the operator's cue, not the member's: it is
+ // the only thing that says when the old key may leave this box's
+ // /etc/carlos/relay-masters. Printed per box, because the region is
+ // done with a key only once every relay there has stopped loading
+ // it (discussion #120).
+ if len(rg.Retiring) > 0 {
+ kids := make([]string, 0, len(rg.Retiring))
+ for kid := range rg.Retiring {
+ kids = append(kids, kid)
+ }
+ sort.Strings(kids)
+ parts := make([]string, 0, len(kids))
+ for _, kid := range kids {
+ parts = append(parts, kid+" after "+rg.Retiring[kid])
+ }
+ line += ", retiring " + strings.Join(parts, ",")
+ }
switch {
case rg.Stale:
// STALE is the important word and it gets the warning colour: a
diff --git a/relay_test.go b/relay_test.go
index 8fc848383..3d98f753f 100644
--- a/relay_test.go
+++ b/relay_test.go
@@ -170,6 +170,25 @@ func TestPrintRelayStatus(t *testing.T) {
// an old record — the two words a member acts on.
wantNot: []string{"STALE", "older record", "disable pending"},
},
+ {
+ // The operator's mid-rotation question, and the only place the
+ // CLI can answer it: when may the old key leave this box's
+ // /etc/carlos/relay-masters (discussion #120).
+ name: "a retiring master names when it may be dropped",
+ res: healthy,
+ modifier: func(r *apiRelayReply) {
+ r.RegionStatus = []apiRelayRegion{{
+ Region: "eu-west-1", Box: "relay-1", HeartbeatAge: "12s", Current: true,
+ LeaseLive: true, Listeners: []string{"udp:3478"}, Kids: []string{"k1", "k2"},
+ Retiring: map[string]string{"k2": "2026-09-10T12:00:00Z", "k1": "2026-09-09T12:00:00Z"},
+ }}
+ },
+ // Two of them, deliberately seeded out of order: the kids are
+ // sorted, and the retiring clause uses the same tight separator
+ // the masters list does, so its items cannot be misread as
+ // further clauses of the line.
+ want: []string{"masters k1,k2", "retiring k1 after 2026-09-09T12:00:00Z,k2 after 2026-09-10T12:00:00Z"},
+ },
{
name: "stale names the relay's own reason",
res: healthy,
diff --git a/routes_cli.go b/routes_cli.go
index e6ed64469..4623fe8bf 100644
--- a/routes_cli.go
+++ b/routes_cli.go
@@ -58,13 +58,13 @@ func runRoutes(args []string) {
}
shown := 0
w := newTable(stdout)
- fmt.Fprintln(w, "HOST\tKIND\tUPSTREAM\tCHANNEL\tENVIRONMENT\tREPLICATION\tSTATUS\tHEALTH")
+ fmt.Fprintln(w, "HOST\tKIND\tUPSTREAM\tCHANNEL\tENVIRONMENT\tREPLICATION\tSTATUS\tHEALTH\tDB")
for _, r := range rows {
if *match != "" && !strings.Contains(r.Host, *match) {
continue
}
shown++
- fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
+ fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
r.Host, dashIfEmpty(r.Kind), dashIfEmpty(r.Upstream), dashIfEmpty(r.Channel),
// An empty ENVIRONMENT is a real answer — the route takes the
// app's default bundle — so it prints as a dash rather than a
@@ -79,7 +79,7 @@ func runRoutes(args []string) {
// where "no answer" cost one tenant 97% of a day's replica
// traffic.
replicationCell(r.Replication),
- dashIfEmpty(r.Status), dashIfEmpty(r.Health))
+ dashIfEmpty(r.Status), dashIfEmpty(r.Health), dbCell(r.DBBytes))
// Path routes indented under the hostname whose traffic they split.
// A flipped prefix set has to be visible from the command operators
// already use, or a migration is invisible until something breaks.
@@ -98,7 +98,7 @@ func runRoutes(args []string) {
if p.Host != r.Host {
continue
}
- fmt.Fprintf(w, " /%s\tpath\t%s\t—\t—\t—\t—\t—\n", p.Prefix, dashIfEmpty(p.Target))
+ fmt.Fprintf(w, " /%s\tpath\t%s\t—\t—\t—\t—\t—\t—\n", p.Prefix, dashIfEmpty(p.Target))
}
}
w.Flush()
@@ -131,3 +131,12 @@ func dashIfEmpty(s string) string {
}
return s
}
+
+// dbCell renders a route's database size, a dash (this table's own em dash)
+// when no box has measured it.
+func dbCell(n *int64) string {
+ if n == nil {
+ return "—"
+ }
+ return humanBytes(n)
+}
diff --git a/store.go b/store.go
index 0db093234..cae6a08a1 100644
--- a/store.go
+++ b/store.go
@@ -499,6 +499,21 @@ func printStoreRotate(w io.Writer, res apiStoreReply, finish bool) {
// "provisioned at" — review finding from the walk design: it MOVES on every
// rotate and every drift repair, so a fixed-sounding label would misdescribe
// it the very first time either happens.
+// printStoreSize prints the bucket's size from the console's daily storage
+// sweep, dated, because it is up to a day old: status reads the last sweep
+// rather than listing the bucket on every call.
+func printStoreSize(w io.Writer, sz *apiStoreSize) {
+ switch {
+ case sz == nil:
+ mutedf(w, " size: not measured yet — the console sizes stores once a day")
+ case sz.Unswept:
+ warnf(w, "size: the %s sweep could not read this bucket (credential or endpoint) — no size known", sz.AsOf)
+ default:
+ n := sz.StoreBytes
+ fmt.Fprintf(w, " size: %s (as of %s)\n", humanBytes(&n), sz.AsOf)
+ }
+}
+
func printStoreStatus(w io.Writer, res apiStoreReply) {
c := colorFor(w)
fmt.Fprintf(w, "%s %s\n", c.head("store for"), res.App)
@@ -528,6 +543,7 @@ func printStoreStatus(w io.Writer, res apiStoreReply) {
p := res.Provisioned
fmt.Fprintf(w, " provisioned: bucket %s in %s, user %s, key %s — last credential change %s\n",
p.Bucket, p.Region, p.IAMUser, p.KeyID, p.At.Format(time.RFC3339))
+ printStoreSize(w, res.Size)
} else {
if res.Entitlement.Approved() {
mutedf(w, " nothing is provisioned yet. re-run carlos store create --app %s", res.App)
diff --git a/store_test.go b/store_test.go
index 437596691..f2172ca07 100644
--- a/store_test.go
+++ b/store_test.go
@@ -867,3 +867,29 @@ func TestStoreVerbDispatchAndRefusals(t *testing.T) {
t.Fatalf("bad app name = %q, want the name gate", out)
}
}
+
+// TestPrintStoreStatusSize: a provisioned store prints its swept size with
+// the sweep's date, "not measured yet" before any sweep, and a warning (never
+// a 0) when the sweep could not read the bucket.
+func TestPrintStoreStatusSize(t *testing.T) {
+ base := apiStoreReply{App: "keymail", Backend: "aws",
+ Declared: &apiStoreDeclared{EnvPrefix: "CARLOS_STORE"},
+ Provisioned: &apiStoreProvisioned{Bucket: "b1", Region: "eu-west-1", IAMUser: "u1", KeyID: "AKIA1"}}
+ for _, tc := range []struct {
+ name string
+ size *apiStoreSize
+ want string
+ }{
+ {"swept", &apiStoreSize{AsOf: "2026-09-17", StoreBytes: 3 << 30}, "size: 3.0GB (as of 2026-09-17)"},
+ {"never", nil, "size: not measured yet"},
+ {"unswept", &apiStoreSize{AsOf: "2026-09-17", Unswept: true}, "could not read this bucket"},
+ } {
+ var buf bytes.Buffer
+ res := base
+ res.Size = tc.size
+ printStoreStatus(&buf, res)
+ if !strings.Contains(buf.String(), tc.want) {
+ t.Errorf("%s: output missing %q:\n%s", tc.name, tc.want, buf.String())
+ }
+ }
+}