amadan
cmd/moveprobe: coalesce the readiness store check
Anonymous / and /ready traffic now costs at most one store check per 5 s: one check runs at a time on its own deadline, requests that arrive while it runs share its result, and the result answers for 5 s. Previously every request made its own S3 call. The README notes the cache and the one new body (a request that gave up waiting). Final review I3.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+108
−2
cmd/moveprobe/README.md+3 −1cmd/moveprobe/main.go+48 −1cmd/moveprobe/main_test.go+57 −0
diff --git a/cmd/moveprobe/README.md b/cmd/moveprobe/README.md| index e729077aa..8f386216a 100644 |
| --- a/cmd/moveprobe/README.md |
| +++ b/cmd/moveprobe/README.md |
| @@ -65,7 +65,7 @@ curl -H "Authorization: Bearer $MOVEPROBE_SECRET" https://mp1.example.com/report |
| | path | who can call it | what it does | |
| |---|---|---| |
| -| `/` and `/ready` | anyone | 200 when the key file, the platform key, the stored object and the secret all check out, and the worker's writes are landing. Otherwise 500 with every failed check. A move probes `/`. | |
| +| `/` and `/ready` | anyone | 200 when the key file, the platform key, the stored object and the secret all check out, and the worker's writes are landing. Otherwise 500 with every failed check. The store check runs at most once every 5 seconds, and requests in between share its result. A move probes `/`. | |
| | `GET /report` | `MOVEPROBE_SECRET` | JSON: row counts by kind and region with their lowest and highest sequence numbers, the timer fires by name and due time, both key fingerprints, the stored object, whether the secret is set, the worker's pid and failed writes, and the region the app saw. Any failed database read makes it a 500, never a partial report. | |
| | `POST /arm-once?in=<seconds>` | `MOVEPROBE_SECRET` | Arms a once-timer named `once-<n>` that calls `/once` after that many seconds. Answers 503 when the instance has no control socket. | |
| | `POST /tick`, `POST /once` | the platform | Timer fires. Each one writes a row with the timer's name and due time. | |
| @@ -150,6 +150,8 @@ On `/` and `/ready`: |
| (500). |
| - `store: <key> reads back different bytes`: the object came back changed |
| (500). |
| +- `store: <error>`: the request gave up while waiting for a store check |
| + that another request had started (500). |
| - `worker: exited (<how>)`: the worker has exited. `<how>` is how it |
| ended, such as `exit status 1` or `signal: killed`, or `its status pipe |
| closed` when the app saw it go before learning how (500). |
diff --git a/cmd/moveprobe/main.go b/cmd/moveprobe/main.go| index a548d257c..9d07ecea8 100644 |
| --- a/cmd/moveprobe/main.go |
| +++ b/cmd/moveprobe/main.go |
| @@ -77,6 +77,9 @@ const ( |
| // line (one a second) before readiness fails. |
| workerStale = 3 * time.Second |
| storeWait = 10 * time.Second |
| + // storeCacheFor is how long one readiness store check answers for: |
| + // anonymous /ready traffic costs at most one check per window. |
| + storeCacheFor = 5 * time.Second |
| modeNormal = "normal" |
| modeBeforeWrite = "before-write" |
| @@ -215,6 +218,13 @@ type app struct { |
| objKey string |
| objBody string |
| + // The readiness store check: one in flight at a time, its result kept |
| + // for storeCacheFor. |
| + storeMu sync.Mutex |
| + storeRunning chan struct{} // closed when the running check finishes |
| + storeAt time.Time // when the cached result was taken |
| + storeResult error |
| + |
| mu sync.Mutex |
| objPut bool |
| workerPID int |
| @@ -540,6 +550,43 @@ func (a *app) checkStore(ctx context.Context) error { |
| return nil |
| } |
| +// storeVerdict is checkStore for readiness, coalesced: a result younger than |
| +// storeCacheFor is reused, and a request that arrives while a check runs |
| +// waits for that check instead of starting another. The check itself runs |
| +// on its own deadline, so one caller giving up does not cut it short for the |
| +// others. |
| +func (a *app) storeVerdict(ctx context.Context) error { |
| + a.storeMu.Lock() |
| + if !a.storeAt.IsZero() && time.Since(a.storeAt) < storeCacheFor { |
| + err := a.storeResult |
| + a.storeMu.Unlock() |
| + return err |
| + } |
| + if running := a.storeRunning; running != nil { |
| + a.storeMu.Unlock() |
| + select { |
| + case <-running: |
| + case <-ctx.Done(): |
| + return fmt.Errorf("store: %v", ctx.Err()) |
| + } |
| + a.storeMu.Lock() |
| + err := a.storeResult |
| + a.storeMu.Unlock() |
| + return err |
| + } |
| + done := make(chan struct{}) |
| + a.storeRunning = done |
| + a.storeMu.Unlock() |
| + |
| + err := a.checkStore(context.Background()) |
| + |
| + a.storeMu.Lock() |
| + a.storeResult, a.storeAt, a.storeRunning = err, time.Now(), nil |
| + a.storeMu.Unlock() |
| + close(done) |
| + return err |
| +} |
| + |
| // ---- rows --------------------------------------------------------------------- |
| // insertSeq writes the next row of a sequenced kind. The sequence is taken |
| @@ -748,7 +795,7 @@ func (a *app) readiness(ctx context.Context) (int, string) { |
| if os.Getenv("MOVEPROBE_SECRET") == "" { |
| fails = append(fails, "MOVEPROBE_SECRET is not set") |
| } |
| - if err := a.checkStore(ctx); err != nil { |
| + if err := a.storeVerdict(ctx); err != nil { |
| fails = append(fails, err.Error()) |
| } |
| if err := a.checkWorker(); err != nil { |
diff --git a/cmd/moveprobe/main_test.go b/cmd/moveprobe/main_test.go| index ff7624a30..693ecda3c 100644 |
| --- a/cmd/moveprobe/main_test.go |
| +++ b/cmd/moveprobe/main_test.go |
| @@ -108,6 +108,8 @@ type fixture struct { |
| mu sync.Mutex |
| auths []string |
| storeReq atomic.Int64 |
| + // storeDelay slows every store request, in milliseconds. |
| + storeDelay atomic.Int64 |
| // controlConns counts the connections the control socket accepted. |
| controlConns atomic.Int64 |
| starts int |
| @@ -123,6 +125,7 @@ func newFixture(t *testing.T) *fixture { |
| s3h, _ := leasetest.NewS3Server(f.bucket) |
| f.store = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| f.storeReq.Add(1) |
| + time.Sleep(time.Duration(f.storeDelay.Load()) * time.Millisecond) |
| f.mu.Lock() |
| f.auths = append(f.auths, r.Header.Get("Authorization")) |
| f.mu.Unlock() |
| @@ -1293,3 +1296,57 @@ func TestReadyFailsWhenTheWorkerStops(t *testing.T) { |
| p.stop() |
| }) |
| } |
| + |
| +// TestReadyCoalescesTheStoreCheck: anonymous /ready traffic costs at most one |
| +// store check per 5 s, however many requests arrive, and requests that land |
| +// while a check is running share its result. |
| +func TestReadyCoalescesTheStoreCheck(t *testing.T) { |
| + f := newFixture(t) |
| + p := f.start(f.env) |
| + p.wait("/ready", http.StatusOK) |
| + |
| + burst := func() { |
| + t.Helper() |
| + var wg sync.WaitGroup |
| + codes := make(chan int, 50) |
| + for i := 0; i < 50; i++ { |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + code, _, err := p.do(http.MethodGet, "/ready", "", nil) |
| + if err != nil { |
| + code = 0 |
| + } |
| + codes <- code |
| + }() |
| + } |
| + wg.Wait() |
| + close(codes) |
| + for c := range codes { |
| + if c != http.StatusOK { |
| + t.Errorf("a /ready in the burst answered %d", c) |
| + } |
| + } |
| + } |
| + |
| + // A slow store, so the burst lands while a check is in flight. |
| + f.storeDelay.Store(300) |
| + time.Sleep(5100 * time.Millisecond) // past any cached result |
| + n0 := f.storeReq.Load() |
| + burst() |
| + n1 := f.storeReq.Load() |
| + if n1-n0 > 1 { |
| + t.Errorf("50 concurrent /ready calls made %d store requests, want at most 1", n1-n0) |
| + } |
| + burst() |
| + if n2 := f.storeReq.Load(); n2 != n1 { |
| + t.Errorf("a second burst inside the 5 s window made %d store requests, want 0", n2-n1) |
| + } |
| + time.Sleep(5100 * time.Millisecond) |
| + n3 := f.storeReq.Load() |
| + burst() |
| + if n4 := f.storeReq.Load(); n4-n3 > 1 { |
| + t.Errorf("the next window's burst made %d store requests, want at most 1", n4-n3) |
| + } |
| + p.stop() |
| +} |