carlos / platform Public

A request gate in the replica client: a refusal stops the attempt before it is sent or counted

The request meter sends when it cannot record, by design, so a recorder
that refuses could never stop a request (second review, both reviewers,
P1). Permission to send is now its own middleware, outside the meter in
the same step: a refused attempt reaches neither the network nor the
meter, and litestream sees a failed upload. Every metered database must
name a gate. Tested through a real S3 client in both option orders, on a
retry, and through the store's own clients.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 3 weeks ago 137fd40950de5cc8e33ed83e2815ecc1138bc65f parent 9c08ce3

5 files changed, +337 −19

internal/lease/meteredstore.go +17 −3
internal/lease/meteredstore_test.go +18 −14
internal/lease/requestgate.go +76 −0
internal/lease/requestgate_test.go +224 −0
internal/lease/requestmeter.go +2 −2

diff --git a/internal/lease/meteredstore.go b/internal/lease/meteredstore.go

index 367dbbf6a..6533ecb74 100644

--- a/internal/lease/meteredstore.go

+++ b/internal/lease/meteredstore.go

@@ -31,7 +31,9 @@ import (

// What it refuses. A database with no meter. There is no unmetered

// fallback here, on purpose: a write that is not counted is exactly what

-// the attestation would be lying about.

+// the attestation would be lying about. And a database with no gate: the

+// meter sends when it cannot record (and holds the day), so stopping a

+// request is the gate's job and not the meter's.

// What it is not, yet. Nothing calls it. It does not replace the external

// binary, change a unit, or attest anything. Replacing the binary also

@@ -49,6 +51,12 @@ type MeteredDB struct {

// Meter records every request the replica client makes. It is bound

// by the caller to the database's owner. Required.

Meter RequestRecorder

+ // Gate is asked before every request the replica client makes, and a

+ // refusal stops the request before it is sent (RequestGate). It is

+ // where the caller says whether this database is still the owner's to

+ // replicate. Required: a client nothing can stop would go on writing,

+ // uncounted, after its meter had been told to stop counting.

+ Gate RequestGate

// MetaDir is where litestream keeps this database's shadow state.

// Required, absolute, and not shared with any other database. The

// caller puts it in a tree only the platform can write: beside the

@@ -63,7 +71,7 @@ type MeteredDB struct {

// NewMeteredStore builds one litestream store over dbs, each replica client

// metered by its own recorder. The store is returned unopened. It refuses

-// an empty list, a database with no path, prefix or meter, and two

+// an empty list, a database with no path, prefix, meter or gate, and two

// databases that share a file or a replica prefix (two owners must never

// be billed for one stream, nor one stream written twice).

func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store, error) {

@@ -98,6 +106,8 @@ func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store,

return nil, fmt.Errorf("lease: metered store: %s has no replica prefix", d.Path)

case d.Meter == nil:

return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, errRequestMeterRequired)

+ case d.Gate == nil:

+ return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, errRequestGateRequired)

case paths[filepath.Clean(d.Path)]:

return nil, fmt.Errorf("lease: metered store: %s is listed twice", d.Path)

case prefixes[d.Prefix]:

@@ -115,7 +125,11 @@ func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store,

// caller's, in a tree tenants cannot write (MeteredDB.MetaDir).

db.SetMetaPath(filepath.Clean(d.MetaDir))

db.MonitorInterval = dbMonitorInterval

- db.Replica = litestream.NewReplicaWithClient(db, newReplicaClient(bucket, d.Prefix, region, []func(*awss3.Options){option}...))

+ gate, err := requestGateOption(d.Gate)

+ if err != nil {

+ return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, err)

+ }

+ db.Replica = litestream.NewReplicaWithClient(db, newReplicaClient(bucket, d.Prefix, region, []func(*awss3.Options){gate, option}...))

if d.SyncInterval > 0 {

db.Replica.SyncInterval = d.SyncInterval

}

diff --git a/internal/lease/meteredstore_test.go b/internal/lease/meteredstore_test.go

index bfd400823..a78abe152 100644

--- a/internal/lease/meteredstore_test.go

+++ b/internal/lease/meteredstore_test.go

@@ -20,8 +20,8 @@ func TestMeteredStoreCountsEveryRequestOnItsOwnersMeter(t *testing.T) {

dir := t.TempDir()

one, two := &memoryRequestRecorder{}, &memoryRequestRecorder{}

store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{

- {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, MetaDir: filepath.Join(dir, ".replicator", "one")},

- {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two, MetaDir: filepath.Join(dir, ".replicator", "two")},

+ {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, Gate: admitEveryRequest, MetaDir: filepath.Join(dir, ".replicator", "one")},

+ {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two, Gate: admitEveryRequest, MetaDir: filepath.Join(dir, ".replicator", "two")},

})

if err != nil {

t.Fatal(err)

@@ -66,7 +66,7 @@ func TestMeteredStoreRefusesWhatItCannotAttribute(t *testing.T) {

dir := t.TempDir()

m := &memoryRequestRecorder{}

metaA, metaB := filepath.Join(dir, ".replicator", "a"), filepath.Join(dir, ".replicator", "b")

- ok := MeteredDB{Path: filepath.Join(dir, "a.db"), Prefix: "replicas/a", Meter: m, MetaDir: metaA}

+ ok := MeteredDB{Path: filepath.Join(dir, "a.db"), Prefix: "replicas/a", Meter: m, Gate: admitEveryRequest, MetaDir: metaA}

// The same file under another name: a link a tenant could plant.

if err := os.WriteFile(ok.Path, []byte("x"), 0o644); err != nil {

t.Fatal(err)

@@ -77,24 +77,28 @@ func TestMeteredStoreRefusesWhatItCannotAttribute(t *testing.T) {

}

for name, dbs := range map[string][]MeteredDB{

"no databases": nil,

- "no meter": {{Path: ok.Path, Prefix: ok.Prefix, MetaDir: metaA}},

- "no prefix": {{Path: ok.Path, Meter: m, MetaDir: metaA}},

- "a relative path": {{Path: "a.db", Prefix: ok.Prefix, Meter: m, MetaDir: metaA}},

- "no metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m}},

- "a relative metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m, MetaDir: ".replicator/a"}},

- "one metadata directory for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", Meter: m, MetaDir: metaA}},

- "one file listed twice": {ok, {Path: filepath.Join(dir, ".", "a.db"), Prefix: "replicas/b", Meter: m, MetaDir: metaB}},

- "one file under a link": {ok, {Path: link, Prefix: "replicas/b", Meter: m, MetaDir: metaB}},

- "one prefix for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: ok.Prefix, Meter: m, MetaDir: metaB}},

- "a meter missing in many": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", MetaDir: metaB}},

+ "no meter": {{Path: ok.Path, Prefix: ok.Prefix, Gate: admitEveryRequest, MetaDir: metaA}},

+ "no gate": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m, MetaDir: metaA}},

+ "no prefix": {{Path: ok.Path, Meter: m, Gate: admitEveryRequest, MetaDir: metaA}},

+ "a relative path": {{Path: "a.db", Prefix: ok.Prefix, Meter: m, Gate: admitEveryRequest, MetaDir: metaA}},

+ "no metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m, Gate: admitEveryRequest}},

+ "a relative metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m, Gate: admitEveryRequest, MetaDir: ".replicator/a"}},

+ "one metadata directory for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", Meter: m, Gate: admitEveryRequest, MetaDir: metaA}},

+ "one file listed twice": {ok, {Path: filepath.Join(dir, ".", "a.db"), Prefix: "replicas/b", Meter: m, Gate: admitEveryRequest, MetaDir: metaB}},

+ "one file under a link": {ok, {Path: link, Prefix: "replicas/b", Meter: m, Gate: admitEveryRequest, MetaDir: metaB}},

+ "one prefix for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: ok.Prefix, Meter: m, Gate: admitEveryRequest, MetaDir: metaB}},

+ "a meter missing in many": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", Gate: admitEveryRequest, MetaDir: metaB}},

} {

if store, err := NewMeteredStore("test-bucket", "eu-west-1", dbs); err == nil || store != nil {

t.Errorf("%s: accepted", name)

}

- if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix, MetaDir: metaA}}); !errors.Is(err, errRequestMeterRequired) {

+ if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix, Gate: admitEveryRequest, MetaDir: metaA}}); !errors.Is(err, errRequestMeterRequired) {

t.Errorf("no meter: %v, want the meter-required error", err)

}

+ if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix, Meter: m, MetaDir: metaA}}); !errors.Is(err, errRequestGateRequired) {

+ t.Errorf("no gate: %v, want the gate-required error", err)

+ }

if _, err := NewMeteredStore("", "eu-west-1", []MeteredDB{ok}); err == nil {

t.Error("no bucket: accepted")

}

diff --git a/internal/lease/requestgate.go b/internal/lease/requestgate.go

new file mode 100644

index 000000000..6b3bfe345

--- /dev/null

+++ b/internal/lease/requestgate.go

@@ -0,0 +1,76 @@

+package lease

+import (

+ "context"

+ "errors"

+ "fmt"

+ "github.com/aws/aws-sdk-go-v2/service/s3"

+ "github.com/aws/smithy-go/middleware"

+// RequestGate decides whether one physical attempt may be sent. It is asked

+// on the request path, once for every attempt (a retry is asked again),

+// before the attempt is recorded and before it is dispatched. A non-nil

+// answer stops the attempt: nothing is written to the network, nothing is

+// recorded, and the caller of the S3 operation gets ErrRequestRefused

+// wrapping the gate's own error.

+//

+// It exists apart from the request meter on purpose. The meter records; a

+// recording that fails must not stop replication, so the meter sends anyway

+// and holds the day. Permission to send is a different question, and an

+// answer of "no" has to reach litestream as a failed upload so that it does

+// not advance its replica position.

+//

+// The gate is given the request's context and decides for itself how long

+// it is willing to take. An attempt waiting on its gate has not been sent.

+type RequestGate func(ctx context.Context) error

+// ErrRequestRefused: a request gate stopped an attempt before dispatch.

+var ErrRequestRefused = errors.New("lease: replica request refused before dispatch")

+var errRequestGateRequired = errors.New("replication_gate_required")

+const (

+ requestGateMiddlewareID = "CarlosReplicationRequestGate"

+ requestMeterMiddlewareID = "CarlosReplicationRequestMeter"

+// requestGateOption puts gate in front of every attempt an S3 client makes.

+// It sits in the same step as the request meter and outside it, whichever

+// of the two options is applied first: a refused attempt never reaches the

+// meter, so it is neither counted nor left as a started row nobody

+// finishes. A second gate on one client is a wiring error and fails every

+// operation, rather than letting either gate think it is the only one.

+func requestGateOption(gate RequestGate) (func(*s3.Options), error) {

+ if gate == nil {

+ return nil, errRequestGateRequired

+ }

+ return func(options *s3.Options) {

+ options.APIOptions = append(options.APIOptions, func(stack *middleware.Stack) error {

+ if _, exists := stack.Deserialize.Get(requestGateMiddlewareID); exists {

+ return fmt.Errorf("%w: two request gates on one client", ErrRequestRefused)

+ }

+ hook := middleware.DeserializeMiddlewareFunc(requestGateMiddlewareID, func(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (middleware.DeserializeOutput, middleware.Metadata, error) {

+ if err := askRequestGate(ctx, gate); err != nil {

+ return middleware.DeserializeOutput{}, middleware.Metadata{}, fmt.Errorf("%w: %w", ErrRequestRefused, err)

+ }

+ return next.HandleDeserialize(ctx, in)

+ })

+ if _, metered := stack.Deserialize.Get(requestMeterMiddlewareID); metered {

+ return stack.Deserialize.Insert(hook, requestMeterMiddlewareID, middleware.Before)

+ }

+ return stack.Deserialize.Add(hook, middleware.After)

+ })

+ }, nil

+// askRequestGate asks the gate, and reads a gate that panics as a refusal.

+func askRequestGate(ctx context.Context, gate RequestGate) (err error) {

+ defer func() {

+ if recover() != nil {

+ err = errors.New("request gate panicked")

+ }

+ }()

+ return gate(ctx)

diff --git a/internal/lease/requestgate_test.go b/internal/lease/requestgate_test.go

new file mode 100644

index 000000000..c358ee62b

--- /dev/null

+++ b/internal/lease/requestgate_test.go

@@ -0,0 +1,224 @@

+package lease

+import (

+ "bytes"

+ "context"

+ "errors"

+ "net/http"

+ "net/http/httptest"

+ "path/filepath"

+ "sync/atomic"

+ "testing"

+ "time"

+ "github.com/aws/aws-sdk-go-v2/aws"

+ "github.com/aws/aws-sdk-go-v2/aws/retry"

+ "github.com/aws/aws-sdk-go-v2/credentials"

+ "github.com/aws/aws-sdk-go-v2/service/s3"

+func admitEveryRequest(context.Context) error { return nil }

+// gatedTestClient is a real S3 client over a test server with a gate and a

+// 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)

+ }

+ gateOpt, err := requestGateOption(gate)

+ if err != nil {

+ t.Fatal(err)

+ }

+ cfg := aws.Config{Region: "eu-west-1", Credentials: credentials.NewStaticCredentialsProvider("dummy", "dummy", ""), HTTPClient: srv.Client(), Retryer: func() aws.Retryer {

+ return retry.AddWithMaxBackoffDelay(retry.NewStandard(func(o *retry.StandardOptions) { o.MaxAttempts = 3 }), time.Millisecond)

+ }}

+ base := func(o *s3.Options) {

+ o.BaseEndpoint = aws.String(srv.URL)

+ o.UsePathStyle = true

+ o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired

+ o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired

+ }

+ if gateFirst {

+ return s3.NewFromConfig(cfg, base, gateOpt, meterOpt)

+ }

+ return s3.NewFromConfig(cfg, base, meterOpt, gateOpt)

+// A refusal stops the attempt: nothing reaches the server, nothing is

+// recorded (not even a started row or a held day), the caller gets the

+// gate's error, and the SDK does not try again. This is the difference

+// from the meter, which sends when it cannot record. It holds whichever of

+// the two options the client was given first.

+func TestRequestGateStopsAnAttemptBeforeItIsSentOrCounted(t *testing.T) {

+ for name, gateFirst := range map[string]bool{"gate option first": true, "meter option first": false} {

+ t.Run(name, func(t *testing.T) {

+ recorder := &memoryRequestRecorder{}

+ var received, asked atomic.Int32

+ var refuse atomic.Bool

+ stale := errors.New("not the owner's any more")

+ refuse.Store(true)

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {

+ received.Add(1)

+ w.WriteHeader(http.StatusOK)

+ }, func(context.Context) error {

+ asked.Add(1)

+ if refuse.Load() {

+ return stale

+ }

+ return nil

+ }, recorder, gateFirst)

+ err := meterPut(context.Background(), client)

+ if !errors.Is(err, ErrRequestRefused) || !errors.Is(err, stale) {

+ t.Fatalf("a refused request returned %v, want ErrRequestRefused wrapping the gate's error", err)

+ }

+ if received.Load() != 0 {

+ t.Fatalf("%d requests reached the server after a refusal", received.Load())

+ }

+ if asked.Load() != 1 {

+ t.Fatalf("the gate was asked %d times for one refused call: a refusal must not be retried", asked.Load())

+ }

+ recorder.mu.Lock()

+ if len(recorder.requests) != 0 || recorder.incomplete != 0 || recorder.holds != 0 {

+ t.Fatalf("a refused request reached the meter: %d started, %d incomplete, %d holds", len(recorder.requests), recorder.incomplete, recorder.holds)

+ }

+ recorder.mu.Unlock()

+ // The same client, admitted: sent once and counted once.

+ refuse.Store(false)

+ if err := meterPut(context.Background(), client); err != nil {

+ t.Fatal(err)

+ }

+ recorder.mu.Lock()

+ defer recorder.mu.Unlock()

+ if received.Load() != 1 || len(recorder.requests) != 1 || !recorder.requests[0].finished {

+ t.Fatalf("an admitted request: %d reached the server, %d metered", received.Load(), len(recorder.requests))

+ }

+ })

+ }

+// The gate is asked for every physical attempt, not once for the

+// operation: a retry the SDK makes after the gate has closed is not sent.

+func TestRequestGateIsAskedAgainForARetry(t *testing.T) {

+ recorder := &memoryRequestRecorder{}

+ var received, asked atomic.Int32

+ var refuse atomic.Bool

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {

+ received.Add(1)

+ // The first attempt fails in a way the SDK retries, and the

+ // gate closes while it is in flight.

+ refuse.Store(true)

+ w.WriteHeader(http.StatusInternalServerError)

+ _, _ = w.Write([]byte(`<Error><Code>InternalError</Code></Error>`))

+ }, func(context.Context) error {

+ asked.Add(1)

+ if refuse.Load() {

+ return errors.New("closed")

+ }

+ return nil

+ }, recorder, true)

+ err := meterPut(context.Background(), client)

+ if !errors.Is(err, ErrRequestRefused) {

+ t.Fatalf("the retry returned %v, want ErrRequestRefused", err)

+ }

+ if received.Load() != 1 || asked.Load() != 2 {

+ t.Fatalf("%d attempts reached the server and the gate was asked %d times: want 1 and 2", received.Load(), asked.Load())

+ }

+ recorder.mu.Lock()

+ defer recorder.mu.Unlock()

+ if len(recorder.requests) != 1 || !recorder.requests[0].finished {

+ t.Fatalf("the attempt that was sent must be metered whole, and only it: %+v", recorder.requests)

+ }

+// A gate that panics has not said yes.

+func TestRequestGateThatPanicsRefuses(t *testing.T) {

+ recorder := &memoryRequestRecorder{}

+ var received atomic.Int32

+ client := gatedTestClient(t, func(w http.ResponseWriter, r *http.Request) {

+ received.Add(1)

+ w.WriteHeader(http.StatusOK)

+ }, func(context.Context) error { panic("gate") }, recorder, true)

+ if err := meterPut(context.Background(), client); !errors.Is(err, ErrRequestRefused) {

+ t.Fatalf("a panicking gate returned %v", err)

+ }

+ if received.Load() != 0 {

+ t.Fatal("a request was sent past a gate that panicked")

+ }

+// Two gates on one client is a wiring error. Every operation fails and

+// nothing is sent, whatever the gates would have said.

+func TestTwoRequestGatesOnOneClientSendNothing(t *testing.T) {

+ var received atomic.Int32

+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ received.Add(1)

+ w.WriteHeader(http.StatusOK)

+ }))

+ t.Cleanup(srv.Close)

+ one, _ := requestGateOption(admitEveryRequest)

+ two, _ := requestGateOption(admitEveryRequest)

+ cfg := aws.Config{Region: "eu-west-1", Credentials: credentials.NewStaticCredentialsProvider("dummy", "dummy", ""), HTTPClient: srv.Client()}

+ client := s3.NewFromConfig(cfg, func(o *s3.Options) {

+ o.BaseEndpoint = aws.String(srv.URL)

+ o.UsePathStyle = true

+ }, one, two)

+ if err := meterPut(context.Background(), client); !errors.Is(err, ErrRequestRefused) {

+ t.Fatalf("two gates: %v, want ErrRequestRefused", err)

+ }

+ if received.Load() != 0 {

+ t.Fatal("a request was sent through a client with two gates")

+ }

+ if _, err := requestGateOption(nil); !errors.Is(err, errRequestGateRequired) {

+ t.Fatalf("no gate: %v", err)

+ }

+// The store's own clients, the ones litestream writes through: a database

+// whose gate refuses sends nothing and counts nothing, litestream's write

+// fails, and its neighbour in the same store is untouched.

+func TestMeteredStoreClientIsStoppedByItsGate(t *testing.T) {

+ f, endpoint := newFakeS3(t)

+ fakeCreds(t)

+ t.Setenv("CARLOS_S3_ENDPOINT", endpoint)

+ dir := t.TempDir()

+ one, two := &memoryRequestRecorder{}, &memoryRequestRecorder{}

+ var closed atomic.Bool

+ store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{

+ {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, MetaDir: filepath.Join(dir, ".replicator", "one"), Gate: func(context.Context) error {

+ if closed.Load() {

+ return errors.New("closed")

+ }

+ return nil

+ }},

+ {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two, MetaDir: filepath.Join(dir, ".replicator", "two"), Gate: admitEveryRequest},

+ })

+ if err != nil {

+ t.Fatal(err)

+ }

+ dbs := store.DBs()

+ ctx := context.Background()

+ if _, err := dbs[0].Replica.Client.WriteLTXFile(ctx, 0, 2, 2, bytes.NewReader(mustLTXBytes(t))); err != nil {

+ t.Fatal(err)

+ }

+ closed.Store(true)

+ if _, err := dbs[0].Replica.Client.WriteLTXFile(ctx, 0, 3, 3, bytes.NewReader(mustLTXBytes(t))); !errors.Is(err, ErrRequestRefused) {

+ t.Fatalf("litestream's write through a closed gate returned %v, want ErrRequestRefused", err)

+ }

+ if _, err := dbs[1].Replica.Client.WriteLTXFile(ctx, 0, 2, 2, bytes.NewReader(mustLTXBytes(t))); err != nil {

+ t.Fatalf("the neighbour's write: %v", err)

+ }

+ one.mu.Lock()

+ defer one.mu.Unlock()

+ two.mu.Lock()

+ defer two.mu.Unlock()

+ if len(one.requests) != 1 || len(two.requests) != 1 {

+ t.Fatalf("meters saw %d and %d requests, want 1 and 1", len(one.requests), len(two.requests))

+ }

+ if got := len(f.requests()); got != 2 {

+ t.Fatalf("%d requests reached the bucket, want 2: %v", got, f.requests())

+ }

diff --git a/internal/lease/requestmeter.go b/internal/lease/requestmeter.go

index 620b49e34..ac4baf2a2 100644

--- a/internal/lease/requestmeter.go

+++ b/internal/lease/requestmeter.go

@@ -76,7 +76,7 @@ func requestMeterOption(recorder requestRecorder) (func(*s3.Options), error) {

slots := make(chan struct{}, requestMeterSlots)

return func(options *s3.Options) {

options.APIOptions = append(options.APIOptions, func(stack *middleware.Stack) error {

- if previous, exists := stack.Deserialize.Get("CarlosReplicationRequestMeter"); exists {

+ if previous, exists := stack.Deserialize.Get(requestMeterMiddlewareID); exists {

now := time.Now().UTC()

if hook, ok := previous.(*requestMeterHook); ok {

hook.recorder.Incomplete(now)

@@ -84,7 +84,7 @@ func requestMeterOption(recorder requestRecorder) (func(*s3.Options), error) {

recorder.Incomplete(now)

return nil

}

- return stack.Deserialize.Add(&requestMeterHook{recorder: recorder, DeserializeMiddleware: middleware.DeserializeMiddlewareFunc("CarlosReplicationRequestMeter", func(ctx context.Context, in middleware.DeserializeInput, next middleware.DeserializeHandler) (middleware.DeserializeOutput, middleware.Metadata, error) {

+ 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)

})}, middleware.After)

})