merge

Taskfile.yml

@@ -1,7 +1,11 @@
-version: '3'
+version: "3"
 
 tasks:
+  pull:
+    cmds:
+      - docker compose -f ./deployment/docker-compose.yaml pull
   build:
+    deps: [pull]
     cmds:
       - docker compose -f ./deployment/docker-compose.yaml build
 
@@ -13,7 +17,6 @@ tasks:
     cmds:
       - docker compose -f ./deployment/docker-compose.yaml up -d
 
-
   migrate:
     cmds:
       - task: migrate-db
@@ -34,7 +37,6 @@ tasks:
     cmds:
       - goose down-to 0
 
-
   migrate-db:
     env:
       DRIZZLE_DATABASE_URL: "mysql://unkey:password@localhost:3306/unkey"
@@ -49,7 +51,6 @@ tasks:
       - task: seed
       - pnpm test:integration
 
-
   generate-sql:
     dir: internal/db
     cmds:

apps/agent/services/ratelimit/bucket.go

@@ -42,18 +42,18 @@ func (b bucketKey) toString() string {
 // getBucket returns a bucket for the given key and will create one if it does not exist.
 // It returns the bucket and a boolean indicating if the bucket existed before.
 func (s *service) getBucket(key bucketKey) (*bucket, bool) {
-	s.bucketsLock.RLock()
+	s.bucketsMu.RLock()
 	b, ok := s.buckets[key.toString()]
-	s.bucketsLock.RUnlock()
+	s.bucketsMu.RUnlock()
 	if !ok {
 		b = &bucket{
 			limit:    key.limit,
 			duration: key.duration,
 			windows:  make(map[int64]*ratelimitv1.Window),
 		}
-		s.bucketsLock.Lock()
+		s.bucketsMu.Lock()
 		s.buckets[key.toString()] = b
-		s.bucketsLock.Unlock()
+		s.bucketsMu.Unlock()
 	}
 	return b, ok
 }

apps/agent/services/ratelimit/peer.go

@@ -76,7 +76,7 @@ type peer struct {
 }
 
 // getAllPeers returns clients for all nodes in the cluster except ourselves
-func (s *service) getAllPeers(ctx context.Context) ([]peer, error) {
+func (s *service) getAllPeers(context.Context) ([]peer, error) {
 	peers := []peer{}
 	for _, p := range s.cluster.Peers() {
 		if p.Id == s.cluster.NodeId() {

apps/agent/services/ratelimit/ratelimit_replication_test.go

@@ -112,9 +112,9 @@ func TestSync(t *testing.T) {
 
 	originIndex := 0
 	for i, n := range nodes {
-		n.srv.bucketsLock.RLock()
+		n.srv.bucketsMu.RLock()
 		buckets := len(n.srv.buckets)
-		n.srv.bucketsLock.RUnlock()
+		n.srv.bucketsMu.RUnlock()
 		t.Logf("node %d: found %d buckets", i, buckets)
 		if buckets > 0 {
 			originIndex = i
@@ -131,9 +131,9 @@ func TestSync(t *testing.T) {
 	}
 
 	require.Eventually(t, func() bool {
-		nodes[originIndex].srv.bucketsLock.RLock()
+		nodes[originIndex].srv.bucketsMu.RLock()
 		bucket, ok := nodes[originIndex].srv.getBucket(key)
-		nodes[originIndex].srv.bucketsLock.RUnlock()
+		nodes[originIndex].srv.bucketsMu.RUnlock()
 		require.True(t, ok)
 		bucket.RLock()
 		window := bucket.getCurrentWindow(now)

apps/agent/services/ratelimit/service.go

@@ -31,7 +31,7 @@ type service struct {
 
 	shutdownCh chan struct{}
 
-	bucketsLock sync.RWMutex
+	bucketsMu sync.RWMutex
 	// identifier+sequence -> bucket
 	buckets             map[string]*bucket
 	leaseIdToKeyMapLock sync.RWMutex
@@ -61,7 +61,7 @@ func New(cfg Config) (*service, error) {
 		syncBuffer:          nil,
 		mitigateBuffer:      nil,
 		shutdownCh:          make(chan struct{}),
-		bucketsLock:         sync.RWMutex{},
+		bucketsMu:           sync.RWMutex{},
 		buckets:             make(map[string]*bucket),
 		leaseIdToKeyMapLock: sync.RWMutex{},
 		leaseIdToKeyMap:     make(map[string]string),

apps/agent/services/ratelimit/sliding_window.go

@@ -60,8 +60,8 @@ type commitLeaseRequest struct {
 // removeExpiredIdentifiers removes buckets that are no longer relevant
 // for ratelimit decisions
 func (r *service) removeExpiredIdentifiers() {
-	r.bucketsLock.Lock()
-	defer r.bucketsLock.Unlock()
+	r.bucketsMu.Lock()
+	defer r.bucketsMu.Unlock()
 
 	activeRatelimits.Set(float64(len(r.buckets)))
 	now := time.Now()
@@ -260,8 +260,8 @@ func (r *service) SetCounter(ctx context.Context, requests ...setCounterRequest)
 // 		return nil
 // 	}
 
-// 	r.bucketsLock.Lock()
-// 	defer r.bucketsLock.Unlock()
+// 	r.bucketsMu.Lock()
+// 	defer r.bucketsMu.Unlock()
 // 	window, ok := r.buckets[key]
 // 	if !ok {
 // 		r.logger.Warn().Str("key", key).Msg("key not found")

apps/agent/services/ratelimit/sliding_window_test.go

@@ -41,9 +41,9 @@ func TestTakeCreatesWindows(t *testing.T) {
 	require.Equal(t, int64(0), res.previousWindow.Counter)
 	require.Equal(t, int64(1), res.currentWindow.Counter)
 
-	rl.bucketsLock.RLock()
+	rl.bucketsMu.RLock()
 	bucket, ok := rl.buckets[bucketKey{identifier, limit, duration}.toString()]
-	rl.bucketsLock.RUnlock()
+	rl.bucketsMu.RUnlock()
 	require.True(t, ok)
 
 	bucket.Lock()