amadan

carlos / platform Public

Merge branch 'fix-edge-boot-refresh' into main

claude 1b48ed5ebd4e5f8c16557efa1429e76a8c67b612 parent 203f67aparent f39cd37
2 files changed, +198 −5
  • internal/edge/certsync.go +98 −5
  • internal/edge/certsync_test.go +100 −0

CI passed — run details

diff --git a/internal/edge/certsync.go b/internal/edge/certsync.go
index 46c318f3a..021f7bab4 100644
--- a/internal/edge/certsync.go
+++ b/internal/edge/certsync.go
@@ -420,6 +420,19 @@ var certMirrorTimeout = 2 * time.Minute
// serve from while the loop catches up.
var certMirrorBootTimeout = 20 * time.Second
+// certRefreshConcurrency bounds the in-flight gets in Refresh. It is a
+// latency fix, not a throughput one: the pass costs
+// ceil(count/concurrency) x latency instead of count x latency, which at
+// 101 certs and a 250ms round trip is about 2s instead of 25s.
+//
+// Ten, not sixteen, because the AWS SDK's default transport keeps ten idle
+// connections per host (DefaultHTTPTransportMaxIdleConnsPerHost) and nothing
+// here overrides it. Sixteen in flight means six connections closed at the
+// end of every pass and re-handshaked on the next one, forever, on a 30s
+// poll — about a dozen extra cross-region TLS handshakes a minute for one
+// extra wave's worth of latency.
+const certRefreshConcurrency = 10
+
// certSnapshot is one poll's result, swapped in whole. Immutable once
// published, so readers hold no lock beyond the pointer read.
type certSnapshot struct {
@@ -616,6 +629,30 @@ func (m *certMirror) Refresh(ctx context.Context) error {
next := make(map[string][]byte, len(keys))
var tried, failed int
var lastErr error
+
+ // The eligible names first, then their bodies CONCURRENTLY. The fetch is
+ // one round trip per object and the objects are independent, so doing
+ // them one after another makes the whole pass cost
+ // count x latency — which is fine beside the bucket and is not fine
+ // across the planet.
+ //
+ // That is not hypothetical (amadan #118 thread, diagnosed 2026-09-19).
+ // The Sydney box reads this deployment's bucket in eu-west-1 at ~250ms a
+ // round trip. At 101 published certs the serial pass took ~25s, blowing
+ // certMirrorBootTimeout, and since this refresh runs BEFORE either
+ // listener binds, the edge was not serving until it finished. The
+ // self-update gate waits 20s for the edge to answer on loopback, so that
+ // box failed its gate, rolled back, failed the rollback's gate too, and
+ // took no update at all from 2026-09-07 to 2026-09-19.
+ //
+ // The count is what makes it a fleet problem rather than an AU one: every
+ // box pays count x latency, so each new published cert moves every box
+ // closer to the same cliff. us-east-1 at ~75ms is about a third of the
+ // way there at today's count.
+ type fetch struct {
+ key, name string
+ }
+ var items []fetch
for _, key := range keys {
name, ok := strings.CutPrefix(key, certPrefix)
// publishableCertName is the same filter the publish side applies:
@@ -624,8 +661,64 @@ func (m *certMirror) Refresh(ctx context.Context) error {
if !ok || !publishableCertName(name) {
continue
}
- tried++
- body, _, err := m.s.Get(ctx, key)
+ items = append(items, fetch{key: key, name: name})
+ }
+ tried = len(items)
+
+ type result struct {
+ body []byte
+ err error
+ }
+ results := make([]result, len(items))
+ // A fixed pool pulling indices off a channel, NOT a goroutine per object
+ // parked on a semaphore. The distinction only shows up at scale, and it is
+ // the whole point of having a bound: goroutine-per-object costs one stack
+ // per cert before a single one is read, so a deployment with a million
+ // published certs would spend gigabytes on goroutines that are all waiting
+ // for one of sixteen slots.
+ //
+ // Workers also select on ctx.Done, so a cancelled boot refresh stops
+ // handing out work instead of draining the whole queue through Get. (It
+ // still waits for the gets already in flight: Refresh has never been a
+ // hard bound on its own wall clock, and certMirrorBootTimeout bounds the
+ // requests, not this function's return.)
+ workers := min(certRefreshConcurrency, len(items))
+ idx := make(chan int)
+ var wg sync.WaitGroup
+ for range workers {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for i := range idx {
+ body, _, err := m.s.Get(ctx, items[i].key)
+ results[i] = result{body: body, err: err}
+ }
+ }()
+ }
+ for i := range items {
+ select {
+ case idx <- i:
+ case <-ctx.Done():
+ // Leaves results[i:] zero — a nil body with a nil error, which
+ // the apply loop below reads as a successful empty get. Set the
+ // context's error on the rest so they take the failed-get branch
+ // and carry the previous copy forward instead.
+ for j := i; j < len(items); j++ {
+ results[j] = result{err: ctx.Err()}
+ }
+ close(idx)
+ wg.Wait()
+ goto apply
+ }
+ }
+ close(idx)
+ wg.Wait()
+apply:
+
+ // Applied in key order, not completion order, so a partial poll reports
+ // the same lastErr whichever way the fetches happened to interleave.
+ for i, it := range items {
+ body, err := results[i].body, results[i].err
switch {
case errors.Is(err, os.ErrNotExist):
// Raced a delete between the list and the get — a spent token. Not
@@ -639,12 +732,12 @@ func (m *certMirror) Refresh(ctx context.Context) error {
// the previous copy of this name forward and keep going.
failed++
lastErr = err
- if body, ok := prev.entries[name]; ok {
- next[name] = body
+ if body, ok := prev.entries[it.name]; ok {
+ next[it.name] = body
}
continue
}
- next[name] = body
+ next[it.name] = body
}
if tried > 0 && failed == tried {
// Nothing at all could be read: that is a bucket fault, not an object
diff --git a/internal/edge/certsync_test.go b/internal/edge/certsync_test.go
index 7e830e3c1..0dd51fc51 100644
--- a/internal/edge/certsync_test.go
+++ b/internal/edge/certsync_test.go
@@ -11,6 +11,7 @@ import (
"net/http"
"net/http/httptest"
"os"
+ "runtime"
"strings"
"sync"
"testing"
@@ -1851,3 +1852,102 @@ func TestDeploymentHostIsAllowedOnASingleBoxDeployment(t *testing.T) {
t.Fatal("with an empty home map, deploymentHost must be exactly Allowed")
}
}
+
+// slowStore adds a fixed delay to every Get, standing in for a box that
+// reads the deployment bucket from the other side of the planet.
+type slowStore struct {
+ release.Store
+ delay time.Duration
+ mu sync.Mutex
+ peak int // greatest number of gets in flight at once
+ cur int
+ peakGoroutines int
+}
+
+func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.mu.Lock()
+ s.cur++
+ if s.cur > s.peak {
+ s.peak = s.cur
+ }
+ // Sampled while the queue is deepest, which is what tells a fixed worker
+ // pool from one goroutine per object: the first is flat in the number of
+ // objects, the second is not.
+ if g := runtime.NumGoroutine(); g > s.peakGoroutines {
+ s.peakGoroutines = g
+ }
+ s.mu.Unlock()
+ time.Sleep(s.delay)
+ s.mu.Lock()
+ s.cur--
+ s.mu.Unlock()
+ return s.Store.Get(ctx, key)
+}
+
+// TestCertMirrorRefreshFetchesConcurrently is amadan #118's second half as a
+// test. The mirror's boot refresh runs BEFORE either listener binds, so its
+// cost is the edge's time-to-serve; the self-update gate allows 20s for the
+// edge to answer on loopback. Serially, the pass costs count x latency, and
+// the Sydney box (101 certs, ~250ms to eu-west-1) therefore took ~25s to
+// start serving and failed every gate from 2026-09-07 to 2026-09-19.
+//
+// The property is that the pass does NOT cost count x latency. Asserted as
+// wall-clock against a deliberately slow store, and by the peak number of
+// gets in flight, so a change that quietly re-serialises it is caught by
+// both.
+func TestCertMirrorRefreshFetchesConcurrently(t *testing.T) {
+ const (
+ certs = 40
+ delay = 20 * time.Millisecond
+ )
+ inner := release.NewDirStore(t.TempDir())
+ for i := range certs {
+ putBucket(t, inner, fmt.Sprintf("host%02d.example.com", i), []byte("PEM"))
+ }
+ s := &slowStore{Store: inner, delay: delay}
+ m := newCertMirror(s, nil)
+
+ base := runtime.NumGoroutine()
+ start := time.Now()
+ if err := m.Refresh(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ took := time.Since(start)
+
+ // Serial would be at least certs*delay (800ms).
+ if serial := certs * delay; took > serial/2 {
+ t.Errorf("refresh took %s, serial would be ~%s — the gets are not concurrent", took, serial)
+ }
+ // The wall-clock bound above is weak on its own: it admits any
+ // concurrency of about three or more, so a silent drop from ten to four
+ // would pass it while putting a far-from-bucket box back over the update
+ // gate.
+ //
+ // The floor is a LITERAL, deliberately. Deriving it from
+ // certRefreshConcurrency makes the assertion move with the thing it is
+ // meant to pin, and a change of the constant to 4 then passes — which is
+ // exactly what happened to the first version of this line. Eight is the
+ // arithmetic that matters: the update gate allows 20s, a far box is
+ // ~250ms a round trip, so the pass must not exceed ~80 waves, and at a
+ // few hundred published certs that needs most of ten in flight.
+ const floor = 8
+ if s.peak < floor {
+ t.Errorf("peak gets in flight = %d, want at least %d — too little concurrency to fit the update gate on a far-from-bucket box", s.peak, floor)
+ }
+ if s.peak > certRefreshConcurrency {
+ t.Errorf("peak gets in flight = %d, above the %d bound", s.peak, certRefreshConcurrency)
+ }
+ // And the pool is FIXED: goroutines must not scale with the number of
+ // objects. A goroutine per object costs one stack per cert before a
+ // single one is read, which is the shape that breaks at a million certs
+ // rather than at a hundred.
+ if s.peakGoroutines > base+certRefreshConcurrency+8 {
+ t.Errorf("peak goroutines = %d with %d baseline and %d objects — the pool is not fixed", s.peakGoroutines, base, certs)
+ }
+ // And it still mirrors everything it was asked for.
+ for i := range certs {
+ if _, ok := m.Cert(fmt.Sprintf("host%02d.example.com", i)); !ok {
+ t.Fatalf("host%02d.example.com missing from the mirror", i)
+ }
+ }
+}