amadan
edge: bound the refresh's goroutines, not just its connections
Review findings on the previous commit, and the scaling question behind them: what happens at thousands or millions of certs. The bound was a semaphore, so every object still got a goroutine up front and then parked waiting for a slot. That is fine at a hundred certs and is the wrong shape at a million: one stack per cert, allocated before a single one is read. It is now a fixed pool of workers pulling indices off a channel, so goroutines are flat in the number of objects. The workers also select on ctx.Done, so a cancelled boot refresh stops handing out work instead of draining the whole queue through Get, and the objects it never reached take the failed-get branch and carry their previous copy forward rather than reading as successful empty gets. Concurrency drops from sixteen to ten to match the AWS SDK's default of ten idle connections per host. At sixteen, six connections were closed at the end of every pass and re-handshaked on the next one, forever, on a 30s poll -- a dozen extra cross-region TLS handshakes a minute to save one wave's latency. The test's floor was self-referential: it derived the expected concurrency from certRefreshConcurrency, so changing that constant to 4 moved the assertion with it and passed. Watched it do exactly that. The floor is now a literal 8, tied in the comment to the gate arithmetic it protects, and the test also pins that goroutines do not scale with object count -- the regression above would otherwise be invisible. None of this changes the ceiling. The pass still costs ceil(count/10) x latency and still holds every body in memory, so a far box is back over the update gate at roughly 1300 published certs. Fetching only what changed, not blocking listener bind on the pass, and fetching per-SNI on demand are the three that actually retire it; discussion to follow.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2 files changed,
+84
−21
internal/edge/certsync.go+48 −12internal/edge/certsync_test.go+36 −9
CI passed — run details
diff --git a/internal/edge/certsync.go b/internal/edge/certsync.go| index 05a56970b..021f7bab4 100644 |
| --- a/internal/edge/certsync.go |
| +++ b/internal/edge/certsync.go |
| @@ -423,10 +423,15 @@ 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. Bounded so a |
| -// deployment with thousands of published certs opens a predictable number of |
| -// connections rather than one per object. |
| -const certRefreshConcurrency = 16 |
| +// 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. |
| @@ -665,19 +670,50 @@ func (m *certMirror) Refresh(ctx context.Context) error { |
| err error |
| } |
| results := make([]result, len(items)) |
| - sem := make(chan struct{}, certRefreshConcurrency) |
| + // 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 i, it := range items { |
| + for range workers { |
| wg.Add(1) |
| - go func(i int, key string) { |
| + go func() { |
| defer wg.Done() |
| - sem <- struct{}{} |
| - defer func() { <-sem }() |
| - body, _, err := m.s.Get(ctx, key) |
| - results[i] = result{body: body, err: err} |
| - }(i, it.key) |
| + 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. |
diff --git a/internal/edge/certsync_test.go b/internal/edge/certsync_test.go| index df84756a3..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" |
| @@ -1856,10 +1857,11 @@ func TestDeploymentHostIsAllowedOnASingleBoxDeployment(t *testing.T) { |
| // 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 |
| + 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) { |
| @@ -1868,6 +1870,12 @@ func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) |
| 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() |
| @@ -1899,24 +1907,43 @@ func TestCertMirrorRefreshFetchesConcurrently(t *testing.T) { |
| 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). Concurrency 16 puts it |
| - // near ceil(40/16)*20ms = 60ms; half the serial cost is a wide margin |
| - // that still fails loudly if the loop goes back to one at a time. |
| + // 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) |
| } |
| - if s.peak < 2 { |
| - t.Errorf("peak gets in flight = %d, want more than one", s.peak) |
| + // 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 { |