amadan

carlos / platform Public

Make admission association deadline tests deterministic

claude a1177cde8b261fc49126f55e54c863ecefcf51ba parent 44cdddb
3 files changed, +116 −12
  • internal/lease/requestgate_test.go +7 −2
  • internal/lease/requestmeter.go +19 −7
  • internal/lease/requestmeter_admission_test.go +90 −3
diff --git a/internal/lease/requestgate_test.go b/internal/lease/requestgate_test.go
index f8f0924cd..1ec6ffc2f 100644
--- a/internal/lease/requestgate_test.go
+++ b/internal/lease/requestgate_test.go
@@ -24,12 +24,17 @@ func admitEveryRequest(context.Context) (Admission, error) { return Unadmitted,
// meter on it, the two options applied in the order given.
func gatedTestClient(t *testing.T, handler http.HandlerFunc, gate RequestGate, recorder requestRecorder, gateFirst bool) *s3.Client {
t.Helper()
- srv := httptest.NewServer(handler)
- t.Cleanup(srv.Close)
meterOpt, err := requestMeterOption(recorder)
if err != nil {
t.Fatal(err)
}
+ return gatedTestClientWithMeterOption(t, handler, gate, meterOpt, gateFirst)
+}
+
+func gatedTestClientWithMeterOption(t *testing.T, handler http.HandlerFunc, gate RequestGate, meterOpt func(*s3.Options), gateFirst bool) *s3.Client {
+ t.Helper()
+ srv := httptest.NewServer(handler)
+ t.Cleanup(srv.Close)
gateOpt, err := requestGateOption(gate)
if err != nil {
t.Fatal(err)
diff --git a/internal/lease/requestmeter.go b/internal/lease/requestmeter.go
index f95ce7366..fc6493f38 100644
--- a/internal/lease/requestmeter.go
+++ b/internal/lease/requestmeter.go
@@ -69,7 +69,18 @@ type requestMeterHook struct {
recorder requestRecorder
}
+type requestMeterDeadline func() (<-chan time.Time, func())
+
+func realRequestMeterDeadline() (<-chan time.Time, func()) {
+ timer := time.NewTimer(requestMeterBudget)
+ return timer.C, func() { timer.Stop() }
+}
+
func requestMeterOption(recorder requestRecorder) (func(*s3.Options), error) {
+ return requestMeterOptionWithDeadline(recorder, realRequestMeterDeadline)
+}
+
+func requestMeterOptionWithDeadline(recorder requestRecorder, deadline requestMeterDeadline) (func(*s3.Options), error) {
if recorder == nil {
return nil, errRequestMeterRequired
}
@@ -85,7 +96,7 @@ func requestMeterOption(recorder requestRecorder) (func(*s3.Options), error) {
return nil
}
return stack.Deserialize.Add(&requestMeterHook{recorder: recorder, DeserializeMiddleware: middleware.DeserializeMiddlewareFunc(requestMeterMiddlewareID, func(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (middleware.DeserializeOutput, middleware.Metadata, error) {
- return meterAttempt(ctx, in, next, recorder, slots)
+ return meterAttempt(ctx, in, next, recorder, slots, deadline)
})}, middleware.After)
})
}, nil
@@ -103,6 +114,7 @@ type meteredAttempt struct {
writes atomic.Int32
fails atomic.Int32
sealed atomic.Bool
+ deadline requestMeterDeadline
}
// fail counts a proof failure for this attempt. Until the outcome is
@@ -128,9 +140,9 @@ func (a *meteredAttempt) done() {
}
}
-func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler, recorder requestRecorder, slots chan struct{}) (middleware.DeserializeOutput, middleware.Metadata, error) {
+func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler, recorder requestRecorder, slots chan struct{}, deadline requestMeterDeadline) (middleware.DeserializeOutput, middleware.Metadata, error) {
started := time.Now().UTC()
- a := &meteredAttempt{recorder: recorder, started: started}
+ a := &meteredAttempt{recorder: recorder, started: started, deadline: deadline}
a.release = recorder.Hold(started)
a.refs.Store(1)
defer a.done()
@@ -246,11 +258,11 @@ func meterAttempt(ctx context.Context, in middleware.DeserializeInput, next midd
// no worker ran at all (no slot within the budget), and whatever the
// call was for is the request path's to say.
func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), settle func(token uint64, err error, late bool)) (token uint64, err error, ran bool) {
- timer := time.NewTimer(requestMeterBudget)
- defer timer.Stop()
+ expired, stop := a.deadline()
+ defer stop()
select {
case slots <- struct{}{}:
- case <-timer.C:
+ case <-expired:
return 0, errRequestMeterUnavailable, false
}
type result struct {
@@ -283,7 +295,7 @@ func (a *meteredAttempt) call(slots chan struct{}, fn func() (uint64, error), se
select {
case value := <-done:
return value.token, value.err, true
- case <-timer.C:
+ case <-expired:
if claimed.CompareAndSwap(false, true) {
return 0, errRequestMeterUnavailable, true
}
diff --git a/internal/lease/requestmeter_admission_test.go b/internal/lease/requestmeter_admission_test.go
index f40403fcc..dfea9b4a5 100644
--- a/internal/lease/requestmeter_admission_test.go
+++ b/internal/lease/requestmeter_admission_test.go
@@ -5,6 +5,7 @@ import (
"database/sql"
"net/http"
"path/filepath"
+ "sync"
"testing"
"time"
@@ -65,8 +66,21 @@ func TestRequestMeterAdmissionAssociationDurableBeforeSend(t *testing.T) {
t.Fatal(err)
}
defer db.Close()
+ // This tests durable ordering, independently of disk scheduling. The
+ // request context bounds the fixture; deadline behavior has separate
+ // channel-controlled coverage.
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ meterOpt, err := requestMeterOptionWithDeadline(admissionRecorder{recorder}, func() (<-chan time.Time, func()) {
+ expired := make(chan time.Time)
+ stop := context.AfterFunc(ctx, func() { close(expired) })
+ return expired, func() { stop() }
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
var token string
- client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {
+ client := gatedTestClientWithMeterOption(t, func(w http.ResponseWriter, r *http.Request) {
var n int
err := db.QueryRow(`SELECT COUNT(*) FROM admission_attempts link JOIN attempts a ON a.seq = link.attempt_seq JOIN admissions d ON d.token = link.admission_token WHERE d.token = ? AND a.operation = 'PutObject' AND d.account = 'bac' AND d.app = 'one'`, token).Scan(&n)
if err != nil || n != 1 {
@@ -83,8 +97,8 @@ func TestRequestMeterAdmissionAssociationDurableBeforeSend(t *testing.T) {
return nil, err
}
return journalAdmission{t, l, token}, nil
- }, admissionRecorder{recorder}, gateFirst)
- if err := meterPut(context.Background(), client); err != nil {
+ }, meterOpt, gateFirst)
+ if err := meterPut(ctx, client); err != nil {
t.Fatal(err)
}
row, err := l.ReadAdmission(token)
@@ -121,3 +135,76 @@ func TestRequestMeterMissingAdmissionAssociationHoldsDay(t *testing.T) {
t.Fatalf("unrecorded outcome: %+v %v", a, err)
}
}
+
+type controlledStartRecorder struct {
+ memoryRequestRecorder
+ entered chan struct{}
+ release chan struct{}
+ drained chan struct{}
+}
+
+func (r *controlledStartRecorder) Start(op string, at time.Time) (uint64, error) {
+ close(r.entered)
+ <-r.release
+ return r.memoryRequestRecorder.Start(op, at)
+}
+
+func (r *controlledStartRecorder) Hold(at time.Time) func() {
+ release := r.memoryRequestRecorder.Hold(at)
+ return func() {
+ release()
+ close(r.drained)
+ }
+}
+
+func TestRequestMeterControlledDeadlineAbandonsLateStart(t *testing.T) {
+ r := &controlledStartRecorder{entered: make(chan struct{}), release: make(chan struct{}), drained: make(chan struct{})}
+ expired := make(chan time.Time)
+ option, err := requestMeterOptionWithDeadline(r, func() (<-chan time.Time, func()) {
+ return expired, func() {}
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ client := gatedTestClientWithMeterOption(t, func(w http.ResponseWriter, req *http.Request) {
+ w.WriteHeader(http.StatusOK)
+ }, admitEveryRequest, option, true)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ var releaseOnce sync.Once
+ release := func() { releaseOnce.Do(func() { close(r.release) }) }
+ defer release()
+ result := make(chan error, 1)
+ go func() { result <- meterPut(ctx, client) }()
+ select {
+ case <-r.entered:
+ case <-ctx.Done():
+ t.Fatal("Start did not enter")
+ }
+ close(expired)
+ select {
+ case err := <-result:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-ctx.Done():
+ t.Fatal("expired recorder blocked transport")
+ }
+ r.mu.Lock()
+ premature := r.released != 0 || len(r.abandoned) != 0 || len(r.requests) != 0
+ r.mu.Unlock()
+ if premature {
+ t.Fatal("unfinished worker lost its hold or settled before Start")
+ }
+ release()
+ select {
+ case <-r.drained:
+ case <-ctx.Done():
+ t.Fatal("late worker did not release its hold")
+ }
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if len(r.requests) != 1 || r.requests[0].finished || len(r.abandoned) != 1 || r.abandoned[0] != 0 || r.released != 1 || r.incomplete == 0 {
+ t.Fatalf("late Start outcome: requests=%+v abandoned=%v released=%d incomplete=%d", r.requests, r.abandoned, r.released, r.incomplete)
+ }
+}