feat(workflows): add job_test_agent_local.yaml for testing agent locally

feat(workflows): include test_agent_local job in pr.yaml workflow
fix(circuitbreaker): remove unnecessary log message in preflight function
refactor(ratelimit): add defer statement to unlock bucket in Mitigate function
refactor(ratelimit): add context timeout and change log level to debug in broadcastMitigation function
test(ratelimit): update clusterSize range in ratelimit_mitigation_test.go
test(ratelimit): fix index out of range error in ratelimit_test.go
fix(dashboard): update fullName value in createPlainIssue function

.github/workflows/job_test_agent_local.yaml

@@ -0,0 +1,25 @@
+name: Test Agent Local
+on:
+  workflow_call:
+
+
+
+jobs:
+  test_agent_local:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install
+        uses: ./.github/actions/install
+        with:
+          go: true
+
+
+      - name: Build
+        run: task build
+        working-directory: apps/agent
+
+      - name: Test
+        run: go test -cover -json -timeout=60m -failfast ./pkg/... ./services/... | tparse -all -progress
+        working-directory: apps/agent

.github/workflows/pr.yaml

@@ -20,7 +20,9 @@ jobs:
     name: Test API
     uses: ./.github/workflows/job_test_api_local.yaml
 
-
+  test_agent_local:
+    name: Test Agent Local
+    uses: ./.github/workflows/job_test_agent_local.yaml
   # test_agent_integration:
   #   name: Test Agent Integration
   #   runs-on: ubuntu-latest

apps/agent/pkg/circuitbreaker/lib.go

@@ -169,7 +169,6 @@ func (cb *CB[Res]) preflight(ctx context.Context) error {
 	now := cb.config.clock.Now()
 
 	if now.After(cb.resetCountersAt) {
-		cb.logger.Info().Msg("resetting circuit breaker")
 		cb.requests = 0
 		cb.successes = 0
 		cb.failures = 0

apps/agent/services/ratelimit/mitigate.go

@@ -20,8 +20,8 @@ func (s *service) Mitigate(ctx context.Context, req *ratelimitv1.MitigateRequest
 	duration := time.Duration(req.Duration) * time.Millisecond
 	bucket, _ := s.getBucket(bucketKey{req.Identifier, req.Limit, duration})
 	bucket.Lock()
+	defer bucket.Unlock()
 	bucket.windows[req.Window.GetSequence()] = req.Window
-	bucket.Unlock()
 
 	return &ratelimitv1.MitigateResponse{}, nil
 }
@@ -51,6 +51,8 @@ func (s *service) broadcastMitigation(req mitigateWindowRequest) {
 	}
 	for _, peer := range peers {
 		_, err := s.mitigateCircuitBreaker.Do(ctx, func(innerCtx context.Context) (*connect.Response[ratelimitv1.MitigateResponse], error) {
+			innerCtx, cancel := context.WithTimeout(innerCtx, 10*time.Second)
+			defer cancel()
 			return peer.client.Mitigate(innerCtx, connect.NewRequest(&ratelimitv1.MitigateRequest{
 				Identifier: req.identifier,
 				Limit:      req.limit,
@@ -61,7 +63,7 @@ func (s *service) broadcastMitigation(req mitigateWindowRequest) {
 		if err != nil {
 			s.logger.Err(err).Msg("failed to call mitigate")
 		} else {
-			s.logger.Info().Str("peerId", peer.id).Msg("broadcasted mitigation")
+			s.logger.Debug().Str("peerId", peer.id).Msg("broadcasted mitigation")
 		}
 	}
 }

apps/agent/services/ratelimit/ratelimit_mitigation_test.go

@@ -24,7 +24,7 @@ import (
 
 func TestExceedingTheLimitShouldNotifyAllNodes(t *testing.T) {
 
-	for _, clusterSize := range []int{1, 3, 5, 9, 27} {
+	for _, clusterSize := range []int{1, 3, 5} {
 		t.Run(fmt.Sprintf("Cluster Size %d", clusterSize), func(t *testing.T) {
 			logger := logging.New(nil)
 			clusters := []cluster.Cluster{}

apps/agent/services/ratelimit/ratelimit_replication_test.go

@@ -105,7 +105,7 @@ func TestSync(t *testing.T) {
 	}
 
 	// Figure out who is the origin
-	_, err := nodes[1].srv.Ratelimit(ctx, req)
+	_, err := nodes[0].srv.Ratelimit(ctx, req)
 	require.NoError(t, err)
 
 	time.Sleep(5 * time.Second)
@@ -137,7 +137,6 @@ func TestSync(t *testing.T) {
 		require.True(t, ok)
 		bucket.RLock()
 		window := bucket.getCurrentWindow(now)
-		t.Logf("window on origin: %+v", window)
 		counter := window.Counter
 		bucket.RUnlock()
 

apps/agent/services/ratelimit/ratelimit_test.go

@@ -43,7 +43,7 @@ func TestAccuracy_fixed_time(t *testing.T) {
 				logger := logging.New(nil)
 				serfAddrs := []string{}
 
-				for i := range clusterSize {
+				for i := 0; i < clusterSize; i++ {
 					node := Node{}
 					c, serfAddr, rpcAddr := createCluster(t, fmt.Sprintf("node-%d", i), serfAddrs)
 					serfAddrs = append(serfAddrs, serfAddr)
@@ -94,7 +94,6 @@ func TestAccuracy_fixed_time(t *testing.T) {
 				t.Run(fmt.Sprintf("limit:%d", limit), func(t *testing.T) {
 
 					for _, duration := range []time.Duration{
-						1 * time.Second,
 						10 * time.Second,
 						1 * time.Minute,
 						5 * time.Minute,
@@ -143,14 +142,12 @@ func TestAccuracy_fixed_time(t *testing.T) {
 												}
 											}
 
-											exactLimit := limit * (windows + 1)
-											lower := exactLimit
+											lower := limit * windows
 											// At most 150% + 75% per additional ingress node should pass
 											upper := 1.50 + 1.0*float64(len(ingressNodes)-1)
 
 											require.GreaterOrEqual(t, passed, lower)
-											require.LessOrEqual(t, passed, int64(float64(exactLimit)*upper))
-
+											require.LessOrEqual(t, passed, int64(float64(limit*(windows+1))*upper))
 										})
 									}
 

apps/dashboard/lib/trpc/routers/plain.ts

@@ -38,7 +38,6 @@ export const createPlainIssue = rateLimitedProcedure(ratelimit.create)
     }
 
     const email = user.emailAddresses.at(0)!.emailAddress;
-
     const plainUser = await client.upsertCustomer({
       identifier: {
         emailAddress: email,
@@ -49,14 +48,14 @@ export const createPlainIssue = rateLimitedProcedure(ratelimit.create)
           email: email,
           isVerified: user.emailAddresses.at(0)?.verification?.status === "verified",
         },
-        fullName: user.username ?? "",
+        fullName: user.username ?? user.firstName ?? "none avail",
       },
       onUpdate: {
         email: {
           email: email,
           isVerified: user.emailAddresses.at(0)?.verification?.status === "verified",
         },
-        fullName: { value: user.username ?? "" },
+        fullName: { value: user.username ?? user.firstName ?? "none avail" },
       },
     });
     if (plainUser.error) {
@@ -79,7 +78,10 @@ export const createPlainIssue = rateLimitedProcedure(ratelimit.create)
         uiComponent.row({
           mainContent: [uiComponent.plainText({ text: ctx.tenant.id, color: "MUTED" })],
           asideContent: [
-            uiComponent.copyButton({ value: ctx.tenant.id, tooltip: "Copy Tenant Id" }),
+            uiComponent.copyButton({
+              value: ctx.tenant.id,
+              tooltip: "Copy Tenant Id",
+            }),
           ],
         }),
       ],