amadan

carlos / platform Public

relay: allocs gauge resets, honest heartbeat, store-open retry, restartRelay test (review fix)

Four findings from the Task 8 review.

The allocs gauge only ever went up. Flush carries the gauge across
flushes and evicts an app only when all four counters are zero, so
setting it one id at a time could never lower it: an app whose last call
ended reported a phantom allocation in every sample from then on and was
never evicted — a lie in the numbers the bill is made of and a leak in a
process the unit caps at 256 MiB. Meter.SetAllocsAll takes a census
instead and zeroes every id the census does not name.

The heartbeat's read_at was the wall clock, so a relay whose authority
reads had failed for twenty minutes still looked current. It is now
Lease.Last() — the last SUCCESSFUL read — and a new authority_error field
carries the reason it stopped moving, beside the masters_error that was
already there.

`carlos relay serve` died on a store it could not open. With
Restart=always and systemd's start limit that parks the unit in `failed`
within ten seconds, and STUN, which needs no store at all, goes down with
it until a platform update or a human intervenes. relay.OpenStoreWithRetry
waits it out instead, 1s doubling to 30s, one log line a minute, until the
process is asked to stop. Configuration errors stay fatal: no amount of
waiting fixes a missing public IP.

restartRelay had no test. Three now, on applyFixture: the order after the
edge restart, that a try-restart error still yields a successful apply
(the ordinary case is a box with no masters file, whose unit is inactive
by its own condition), and that the rollback path bounces it too.

Also: the arm() comment no longer claims the plan starts no service;
Listeners() is read off the serve loop's mutex; the UDP and TCP listen
seams default independently; the relay unit's condition, OOM adjustment
and absence of CAP_NET_BIND_SERVICE are pinned by a test; and cli.md gives
the console secret's real shape, <kid>:<key>, not the masters file's line.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 5f646e0b911c37a259f55cd7922272924f89c31b parent 36faac6
10 files changed, +466 −28
  • docs/site/cli.md +8 −4
  • internal/host/host_test.go +28 −0
  • internal/host/systemapply.go +5 −2
  • internal/host/systemapply_test.go +85 −0
  • internal/relay/authority.go +21 −7
  • internal/relay/meter.go +22 −0
  • internal/relay/meter_test.go +68 −0
  • internal/relay/serve.go +71 −13
  • internal/relay/serve_test.go +151 −0
  • relay.go +7 −2
diff --git a/docs/site/cli.md b/docs/site/cli.md
index 8e630b69c..b1c35a30f 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -675,10 +675,14 @@ carlos relay keygen
```
`keygen` prints one line — a key id, a key, and `-` for "no retirement date"
-— for a region's master key. Stage it in `/etc/carlos/relay-masters` on that
-region's boxes, mode 0640, root-owned, group `carlos`, and put the same pair
-in the console's `CARLOS_RELAY_MASTERS_<REGION>` secret. The key is printed
-once and stored nowhere by this command.
+— for a region's master key. Stage that line in `/etc/carlos/relay-masters`
+on the region's boxes, mode 0640, root-owned, group `carlos`. The console
+needs the same key in a different shape: set its
+`CARLOS_RELAY_MASTERS_<REGION>` secret to `<kid>:<key>`, or to a
+comma-separated list, `<kid>:<key>,<kid>:<key>`, while a rotation has two
+live. The region name is uppercased with hyphens as underscores, since
+variable names refuse hyphens. The key is printed once and stored nowhere by
+this command.
`serve` is what `carlos-relay.service` execs. The unit only starts on a box
that has a masters file, so a box with no key never listens; a box with one
diff --git a/internal/host/host_test.go b/internal/host/host_test.go
index 2fd411efa..fcd2383bf 100644
--- a/internal/host/host_test.go
+++ b/internal/host/host_test.go
@@ -105,6 +105,34 @@ func TestBootstrapStartsRelayUnit(t *testing.T) {
}
}
+// TestRelayUnitIsGatedAndLosesFirst pins the three lines that make it safe
+// to enable and start carlos-relay.service on every box in the fleet. The
+// condition is what keeps an unarmed box from ever listening; the OOM
+// adjustment and the absence of the bind capability are what keep the relay
+// from costing the edge anything when they share a box (relay spec §3).
+func TestRelayUnitIsGatedAndLosesFirst(t *testing.T) {
+ b, err := unitsFS.ReadFile("units/" + relayUnitName)
+ if err != nil {
+ t.Fatal(err)
+ }
+ unit := string(b)
+ for _, want := range []string{
+ "ConditionPathExists=/etc/carlos/relay-masters",
+ "OOMScoreAdjust=1000",
+ "MemoryMax=256M",
+ "ExecStart=/opt/carlos/carlos relay serve",
+ "ReadWritePaths=/data/relay /run/carlos",
+ } {
+ if !strings.Contains(unit, want) {
+ t.Errorf("%s: missing %q", relayUnitName, want)
+ }
+ }
+ // 3478 is unprivileged; only the edge may bind below 1024.
+ if strings.Contains(unit, "CAP_NET_BIND_SERVICE") {
+ t.Errorf("%s: must not carry CAP_NET_BIND_SERVICE — 3478 needs no capability", relayUnitName)
+ }
+}
+
func TestBootstrapWritesUnitsToStagingDir(t *testing.T) {
root := t.TempDir()
if err := bootstrap(root, nil); err != nil {
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index 2cb0605ba..ab64a3b35 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -675,8 +675,11 @@ func (a *systemApplier) gateOnce(ctx context.Context, target string) error {
return nil
}
-// arm is systemctlPlan verbatim: enable all, start timers and paths, never
-// a service — which is why apply can call it from inside itself.
+// arm is systemctlPlan verbatim: enable all, start timers and paths, and of
+// the services only carlos-relay, whose ConditionPathExists makes the start
+// a no-op on a box with no region master (relay spec §3). It starts nothing
+// else — in particular not the edge, which is running this very code — which
+// is why apply can call it from inside itself.
func (a *systemApplier) arm(plan bootstrapPlan) error {
for _, args := range systemctlPlan(plan.Enabled) {
if _, err := a.ops.systemctl(args...); err != nil {
diff --git a/internal/host/systemapply_test.go b/internal/host/systemapply_test.go
index a16ebe196..e227ef89f 100644
--- a/internal/host/systemapply_test.go
+++ b/internal/host/systemapply_test.go
@@ -691,6 +691,91 @@ func TestApplyRollsBackWhenTheGateFails(t *testing.T) {
}
}
+// The relay rides the platform binary, so a self-update has to bounce it
+// onto the new build the way it bounces the edge — after the edge, and with
+// try-restart rather than restart, because on a box with no region master
+// the unit is inactive by its own ConditionPathExists and there is nothing
+// to restart (relay spec §3).
+func TestApplyRestartsTheRelayAfterTheEdge(t *testing.T) {
+ f, _ := applyHappy(t)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ joined := strings.Join(f.calls, "\n")
+ edge := strings.Index(joined, "systemctl restart carlos-edge")
+ reset := strings.Index(joined, "systemctl reset-failed carlos-relay")
+ try := strings.Index(joined, "systemctl try-restart carlos-relay")
+ switch {
+ case reset < 0 || try < 0:
+ t.Fatalf("relay never restarted:\n%s", joined)
+ case !(edge < reset && reset < try):
+ t.Fatalf("want restart carlos-edge, then reset-failed then try-restart carlos-relay:\n%s", joined)
+ }
+ // `restart` would start the relay on a box whose masters file the
+ // operator never staged. try-restart is the whole point.
+ if strings.Contains(joined, "systemctl restart carlos-relay") {
+ t.Errorf("relay restarted rather than try-restarted:\n%s", joined)
+ }
+}
+
+// A relay that cannot be restarted is not a failed update. The commonest
+// case is the ordinary one — a box with no masters file, whose unit is
+// inactive and whose try-restart systemd may well answer with an error —
+// and the edge, the binary and the gate are all fine.
+func TestApplyRelayRestartFailureDoesNotFailTheUpdate(t *testing.T) {
+ f, req := applyHappy(t)
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "try-restart" && args[1] == systemRelayUnit {
+ orig(args...) // still recorded
+ return "", errors.New("Unit carlos-relay.service is not loaded")
+ }
+ return orig(args...)
+ }
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatalf("a relay that would not restart failed the whole update: %v", err)
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeApplied || !r.matches("stable", "v2", req.PromotedAt) {
+ t.Fatalf("result = %+v", r)
+ }
+ if !f.called("systemctl try-restart " + systemRelayUnit) {
+ t.Fatalf("try-restart never attempted:\n%s", strings.Join(f.calls, "\n"))
+ }
+}
+
+// And the rollback bounces it too: the box is going back to the previous
+// binary, and leaving the relay on the withdrawn one is the same bug the
+// edge's own rollback restart exists to avoid.
+func TestApplyRollbackRestartsTheRelay(t *testing.T) {
+ f, _ := applyHappy(t)
+ f.edgeUp = false
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "restart" {
+ if b, _ := os.ReadFile(systemBinaryPath(f.a.optDir)); strings.Contains(string(b), "carlos v1") {
+ f.edgeUp = true
+ }
+ }
+ return orig(args...)
+ }
+ if err := f.a.Run(context.Background(), false); err == nil {
+ t.Fatal("a failed gate must exit non-zero")
+ }
+ joined := strings.Join(f.calls, "\n")
+ edge := strings.LastIndex(joined, "systemctl restart carlos-edge")
+ try := strings.LastIndex(joined, "systemctl try-restart carlos-relay")
+ if try < 0 || try < edge {
+ t.Fatalf("the rollback left the relay on the withdrawn build:\n%s", joined)
+ }
+ // Twice over the whole run: once for the apply's restart, once for the
+ // rollback's — not just the apply's, which would pass a LastIndex check
+ // by accident if the rollback did nothing.
+ if n := strings.Count(joined, "systemctl try-restart carlos-relay"); n != 2 {
+ t.Errorf("relay try-restarted %d times, want 2 (apply then rollback):\n%s", n, joined)
+ }
+}
+
func TestApplyRollbackFailedLeavesPrevAndAppliedAlone(t *testing.T) {
f, req := applyHappy(t)
writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "applied.json"), systemApplied{Version: "v1", Channel: "stable", At: f.now}, 0o644)
diff --git a/internal/relay/authority.go b/internal/relay/authority.go
index 74937be56..f827bce06 100644
--- a/internal/relay/authority.go
+++ b/internal/relay/authority.go
@@ -61,16 +61,30 @@ type Lease struct{ last time.Time }
func (l *Lease) Renew(now time.Time) { l.last = now }
func (l *Lease) Live(now time.Time) bool { return !l.last.IsZero() && now.Sub(l.last) <= LeaseMaxAge }
+// Last is the time of the last successful read, zero if there has never
+// been one. It is what the heartbeat reports as read_at: the wall clock
+// would say "just now" on a relay whose every read has failed for twenty
+// minutes, which is the one thing a console watching this record must not
+// be told.
+func (l *Lease) Last() time.Time { return l.last }
+
func HeartbeatKey(box string) string { return "control/relay-ack/" + box + ".json" }
type Heartbeat struct {
- Box string `json:"box"`
- Region string `json:"region"`
- WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none
- ReadAt time.Time `json:"read_at"`
- Listeners []string `json:"listeners"`
- Kids []string `json:"kids"`
- MastersError string `json:"masters_error,omitempty"`
+ Box string `json:"box"`
+ Region string `json:"region"`
+ WrittenAt time.Time `json:"written_at"` // of the authority record held; zero if none
+ // ReadAt is the last SUCCESSFUL authority read — the lease clock, not
+ // the wall clock — so the gap between it and the heartbeat's own
+ // arrival is how far the relay has drifted from the record it serves.
+ ReadAt time.Time `json:"read_at"`
+ Listeners []string `json:"listeners"`
+ Kids []string `json:"kids"`
+ // 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.
+ MastersError string `json:"masters_error,omitempty"`
+ AuthorityError string `json:"authority_error,omitempty"`
}
// WriteHeartbeat writes a box's heartbeat by read-then-CAS: Get the current
diff --git a/internal/relay/meter.go b/internal/relay/meter.go
index 07b4abc9d..6bc67a32c 100644
--- a/internal/relay/meter.go
+++ b/internal/relay/meter.go
@@ -106,6 +106,28 @@ func (m *Meter) SetAllocs(id Identity, n int) {
m.mu.Unlock()
}
+// SetAllocsAll sets the gauge from a complete census of live allocations:
+// every id in live gets its count, and every id the meter is carrying that
+// live does NOT name gets zero.
+//
+// The zeroing half is the point. Flush keeps an app whose gauge is above
+// zero and evicts one whose four counters are all zero, so an app that is
+// only ever set and never cleared reports a phantom allocation in every
+// sample from then on and is never evicted — a slow leak in a process the
+// unit caps at 256 MiB, and a lie in the numbers the bill is made of.
+// Setting the gauge one id at a time cannot do this, because the ids that
+// need zeroing are precisely the ones the caller no longer sees.
+func (m *Meter) SetAllocsAll(live map[Identity]int) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ for id, a := range m.cur {
+ a.Allocs = live[id] // absent from live: zero
+ }
+ for id, n := range live {
+ m.get(id).Allocs = n
+ }
+}
+
// Flush writes one sample of deltas since the previous flush, appended to
// this box's hourly object by read-append-CAS. The relay is the only
// writer of these objects, so the CAS only ever races with itself.
diff --git a/internal/relay/meter_test.go b/internal/relay/meter_test.go
index 2d77a2ad0..03c7813dc 100644
--- a/internal/relay/meter_test.go
+++ b/internal/relay/meter_test.go
@@ -136,3 +136,71 @@ func TestMeterEvictsDeadApps(t *testing.T) {
t.Fatalf("second sample still has dead app abc/docs: %+v", second.Apps)
}
}
+
+// TestSetAllocsAllZeroesDepartedApps is the leak test. SetAllocs alone can
+// only ever raise a gauge, because the ids that need lowering are precisely
+// the ones a census of live allocations no longer names — so an app whose
+// last call ended would report a phantom allocation in every sample from
+// then on and would never be evicted, growing m.cur without bound in a
+// process the unit caps at 256 MiB.
+func TestSetAllocsAllZeroesDepartedApps(t *testing.T) {
+ s := newMemStore(t)
+ now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
+ m := NewMeter("box1", func() time.Time { return now }, nil)
+ a := Identity{"abc", "meet"}
+ b := Identity{"abc", "docs"}
+
+ // Both live and forwarding.
+ m.Add(a, 10, 20, 0)
+ m.Add(b, 30, 40, 0)
+ m.SetAllocsAll(map[Identity]int{a: 1, b: 2})
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+ // b's call ends part way through the next minute: it forwarded some
+ // bytes and then went, so the census no longer names it.
+ m.Add(b, 5, 5, 0)
+ m.SetAllocsAll(map[Identity]int{a: 1})
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+ // And a third flush finds nothing left to say about b at all.
+ m.SetAllocsAll(map[Identity]int{a: 1})
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+
+ body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var c MeterChunk
+ if err := json.Unmarshal(body, &c); err != nil {
+ t.Fatal(err)
+ }
+ if len(c.Samples) != 3 {
+ t.Fatalf("want 3 samples, got %+v", c.Samples)
+ }
+ if got := c.Samples[0].Apps["abc/docs"].Allocs; got != 2 {
+ t.Errorf("first sample abc/docs allocs = %d, want 2", got)
+ }
+ // The gauge came back DOWN. Without the census this would still read 2,
+ // which is the whole finding.
+ if got, ok := c.Samples[1].Apps["abc/docs"]; !ok || got.Allocs != 0 || got.In != 5 {
+ t.Errorf("second sample abc/docs = %+v ok=%v, want its last 5 bytes with allocs 0", got, ok)
+ }
+ if _, ok := c.Samples[2].Apps["abc/docs"]; ok {
+ t.Errorf("third sample still carries the departed app: %+v", c.Samples[2].Apps)
+ }
+ if got := c.Samples[2].Apps["abc/meet"].Allocs; got != 1 {
+ t.Errorf("third sample abc/meet allocs = %d, want the live app kept at 1", got)
+ }
+ // And it is out of the meter's live set, not merely out of the sample:
+ // the gauge that never fell was a leak as much as a lie.
+ m.mu.Lock()
+ n := len(m.cur)
+ m.mu.Unlock()
+ if n != 1 {
+ t.Errorf("meter still carries %d apps, want only the live one", n)
+ }
+}
diff --git a/internal/relay/serve.go b/internal/relay/serve.go
index bd4e82201..f855f51a5 100644
--- a/internal/relay/serve.go
+++ b/internal/relay/serve.go
@@ -1,10 +1,9 @@
// Package relay: serve.go is the process. Everything else in this package
-// binds a socket or parses a record; this is the loop that owns them for
-// the life of
-// `carlos relay serve` (spec §3). It reads the masters file as it changes,
-// the authority record every minute and on every control/serial change,
-// writes a heartbeat on each read, flushes the meter each minute, sweeps
-// expiry, and sheds under the box's memory pressure.
+// binds a socket or parses a record; this is the loop that owns them for the
+// life of `carlos relay serve` (spec §3). It reads the masters file as it
+// changes, the authority record every minute and on every control/serial
+// change, writes a heartbeat on each read, flushes the meter each minute,
+// sweeps expiry, and sheds under the box's memory pressure.
package relay
import (
@@ -53,8 +52,55 @@ var (
meterInterval = time.Minute
expiryInterval = 10 * time.Second
psiInterval = 5 * time.Second
+
+ // The store-open backoff (OpenStoreWithRetry).
+ storeRetryInitial = time.Second
+ storeRetryMax = 30 * time.Second
+ storeRetryLogEvery = time.Minute
)
+// OpenStoreWithRetry calls open until it succeeds or ctx is done, backing
+// off from a second to half a minute and logging at most one line a minute.
+//
+// It exists because the alternative is worse than it looks. The relay's
+// unit is Restart=always, and systemd's default start limit turns five
+// restarts in ten seconds into a unit that sits in `failed` and stays
+// there: a box whose deployment bucket was unreachable for the fifteen
+// seconds around boot would then serve no STUN until a platform update or
+// a human touched it — and STUN, the whole reason this stage exists, needs
+// no store at all. A relay that waits is a relay that comes back on its
+// own. Configuration errors are NOT retried here; they are fatal at the
+// call site, because no amount of waiting fixes a missing public IP.
+func OpenStoreWithRetry(ctx context.Context, open func(context.Context) (release.Store, error), logf func(string, ...any)) (release.Store, error) {
+ if logf == nil {
+ logf = func(string, ...any) {}
+ }
+ delay := storeRetryInitial
+ var lastLog time.Time
+ for attempt := 1; ; attempt++ {
+ s, err := open(ctx)
+ if err == nil {
+ if attempt > 1 {
+ logf("relay: deploy store opened on attempt %d", attempt)
+ }
+ return s, nil
+ }
+ if now := time.Now(); lastLog.IsZero() || now.Sub(lastLog) >= storeRetryLogEvery {
+ logf("relay: deploy store unavailable after %d attempt(s): %v (retrying in %s; STUN is unaffected)", attempt, err, delay)
+ lastLog = now
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(delay):
+ }
+ delay *= 2
+ if delay > storeRetryMax {
+ delay = storeRetryMax
+ }
+ }
+}
+
// Config is everything `carlos relay serve` needs from its box: who it is,
// where it listens from, what it may not forward to, and the store it reads
// its authority from and writes its heartbeat and metering to.
@@ -184,9 +230,16 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
now := time.Now
meter := NewMeter(cfg.Box, now, logf)
+ // Each defaults on its own, so an empty tcp seam can never silently
+ // take the UDP listener down with it (or the reverse): spec §3 says
+ // both listeners, always, and a half-configured pair is a bug rather
+ // than a supported shape. A test seams both or neither.
udp, tcp := cfg.udp, cfg.tcp
- if udp == "" && tcp == "" {
- udp, tcp = relayUDPAddr, relayTCPAddr
+ if udp == "" {
+ udp = relayUDPAddr
+ }
+ if tcp == "" {
+ tcp = relayTCPAddr
}
srv, err := NewServer(ServerConfig{
Region: cfg.Region,
@@ -243,6 +296,7 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
readAuth := func() {
a, _, err := ReadAuthority(ctx, cfg.Store)
t := now()
+ listeners := srv.Listeners() // outside mu: it takes the server's own lock
mu.Lock()
if err == nil {
auth = a
@@ -252,11 +306,14 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
hb := Heartbeat{
Box: cfg.Box,
Region: cfg.Region,
- ReadAt: t,
- Listeners: srv.Listeners(),
+ ReadAt: lease.Last(), // the last SUCCESSFUL read, never the wall clock
+ Listeners: listeners,
Kids: kidsOf(masters),
MastersError: mastersE,
}
+ if err != nil {
+ hb.AuthorityError = err.Error()
+ }
if cur != nil {
hb.WrittenAt = cur.WrittenAt
}
@@ -281,9 +338,10 @@ func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
readAuth()
flush := func(ctx context.Context) {
- for id, n := range srv.LiveByIdentity() {
- meter.SetAllocs(id, n)
- }
+ // A complete census, not one id at a time: an app whose last
+ // allocation ended has to be told its gauge is zero, or it reports
+ // a phantom allocation forever and the meter never evicts it.
+ meter.SetAllocsAll(srv.LiveByIdentity())
if err := meter.Flush(ctx, cfg.Store); err != nil {
logf("relay: meter flush: %v", err)
}
diff --git a/internal/relay/serve_test.go b/internal/relay/serve_test.go
index 95d7e9165..565b42328 100644
--- a/internal/relay/serve_test.go
+++ b/internal/relay/serve_test.go
@@ -239,6 +239,157 @@ func TestServeWritesHeartbeatAndStops(t *testing.T) {
}
}
+// A failed authority read must not look like a healthy one. read_at is the
+// lease clock — the last SUCCESSFUL read — so the gap between it and the
+// heartbeat's arrival is the relay's real drift from the record it serves;
+// a wall clock there would say "just now" for a relay whose every read has
+// failed for twenty minutes, and the console would believe it.
+func TestServeHeartbeatIsHonestAboutAFailedRead(t *testing.T) {
+ inner := newMemStore(t)
+ store := &flakyStore{Store: inner}
+ dir := t.TempDir()
+ masters := filepath.Join(dir, "relay-masters")
+ if err := os.WriteFile(masters, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil {
+ t.Fatal(err)
+ }
+ cfg := Config{
+ Box: "box-honest", Region: "eu-west-1", MastersPath: masters,
+ PublicIP: net.ParseIP("203.0.113.7"), Store: store, Limits: DefaultLimits(),
+ PressurePath: filepath.Join(dir, "none"), udp: "127.0.0.1:0", tcp: "127.0.0.1:0",
+ }
+
+ 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() })
+
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- Serve(ctx, cfg, logf) }()
+
+ // The first read succeeds, so the lease clock is set and honest.
+ good := waitHeartbeat(t, inner, "box-honest")
+ if good.ReadAt.IsZero() || good.AuthorityError != "" {
+ t.Fatalf("first heartbeat = %+v, want a read time and no error", good)
+ }
+ cancel()
+ if err := <-done; err != nil {
+ t.Fatalf("Serve: %v", err)
+ }
+
+ // Now every authority read fails. Serve is stopped, so drive readAuth's
+ // contract directly through a second run whose store is already broken:
+ // the lease clock must not move and the reason must be recorded.
+ store.failGet(AuthorityKey)
+ ctx2, cancel2 := context.WithCancel(context.Background())
+ done2 := make(chan error, 1)
+ go func() { done2 <- Serve(ctx2, cfg, logf) }()
+ deadline := time.Now().Add(10 * time.Second)
+ var hb Heartbeat
+ for {
+ hb = waitHeartbeat(t, inner, "box-honest")
+ if hb.AuthorityError != "" || time.Now().After(deadline) {
+ break
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+ cancel2()
+ <-done2
+
+ if hb.AuthorityError == "" {
+ t.Fatalf("heartbeat = %+v, want the read error recorded", hb)
+ }
+ // A relay that has never read successfully has no lease and no read
+ // time to report — the one thing it must not do is invent one.
+ if !hb.ReadAt.IsZero() {
+ t.Errorf("read_at = %v on a relay whose every read failed, want zero", hb.ReadAt)
+ }
+}
+
+// flakyStore is a release.Store that can be told to fail Get for one key,
+// so a test can watch what the relay says when the bucket stops answering.
+type flakyStore struct {
+ release.Store
+ mu sync.Mutex
+ fail map[string]bool
+}
+
+func (f *flakyStore) failGet(key string) {
+ f.mu.Lock()
+ if f.fail == nil {
+ f.fail = map[string]bool{}
+ }
+ f.fail[key] = true
+ f.mu.Unlock()
+}
+
+func (f *flakyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ f.mu.Lock()
+ bad := f.fail[key]
+ f.mu.Unlock()
+ if bad {
+ return nil, "", errors.New("bucket unreachable")
+ }
+ return f.Store.Get(ctx, key)
+}
+
+// OpenStoreWithRetry waits a store out rather than dying on it: the unit is
+// Restart=always, systemd's start limit would park it in `failed` within
+// ten seconds, and STUN — which needs no store at all — would go down with
+// it.
+func TestOpenStoreWithRetry(t *testing.T) {
+ defer swapRetryTimings(t)()
+ want := newMemStore(t)
+ calls := 0
+ got, err := OpenStoreWithRetry(context.Background(), func(context.Context) (release.Store, error) {
+ calls++
+ if calls < 3 {
+ return nil, errors.New("no credentials yet")
+ }
+ return want, nil
+ }, t.Logf)
+ if err != nil {
+ t.Fatalf("OpenStoreWithRetry: %v", err)
+ }
+ if calls != 3 {
+ t.Errorf("opener called %d times, want 3 (two failures then a success)", calls)
+ }
+ if got != want {
+ t.Errorf("returned a different store than the opener gave")
+ }
+}
+
+// And it gives up the moment systemd asks the process to stop, rather than
+// holding a shutdown open for the length of the backoff.
+func TestOpenStoreWithRetryStopsOnCancel(t *testing.T) {
+ defer swapRetryTimings(t)()
+ ctx, cancel := context.WithCancel(context.Background())
+ calls := 0
+ _, err := OpenStoreWithRetry(ctx, func(context.Context) (release.Store, error) {
+ calls++
+ if calls == 2 {
+ cancel()
+ }
+ return nil, errors.New("still down")
+ }, nil)
+ if !errors.Is(err, context.Canceled) {
+ t.Fatalf("err = %v, want context.Canceled", err)
+ }
+}
+
+func swapRetryTimings(t *testing.T) func() {
+ t.Helper()
+ oi, om, ol := storeRetryInitial, storeRetryMax, storeRetryLogEvery
+ storeRetryInitial, storeRetryMax, storeRetryLogEvery = time.Millisecond, 2*time.Millisecond, time.Millisecond
+ return func() { storeRetryInitial, storeRetryMax, storeRetryLogEvery = oi, om, ol }
+}
+
// Serve refuses to start without the two things it cannot invent: the box
// label its objects are keyed by, and a store to write them to.
func TestServeRefusesIncompleteConfig(t *testing.T) {
diff --git a/relay.go b/relay.go
index 84191f8c6..2042250ea 100644
--- a/relay.go
+++ b/relay.go
@@ -59,9 +59,14 @@ func runRelayServe(_ []string) {
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
defer stop()
- store, err := host.OpenDeployStore(ctx)
+ // A store that will not open is waited out, not died on: the unit is
+ // Restart=always and systemd's start limit would park it in `failed`
+ // within ten seconds, taking STUN — which needs no store — down with
+ // it. Configuration errors below are fatal, because waiting fixes
+ // nothing about a missing public IP or a malformed CIDR.
+ store, err := relay.OpenStoreWithRetry(ctx, host.OpenDeployStore, log.Printf)
if err != nil {
- log.Fatalf("relay serve: %v", err)
+ return // ctx cancelled: systemd asked us to stop before we ever started
}
cfg, err := relay.ConfigFromEnv(ctx, os.Getenv, host.ReadPublicIP, host.ReadPlacementRegion, store)
if err != nil {