fix: add eviction to unbounded rate-limit/cache/lock maps

- RateLimitStep.buckets: lazy cleanup goroutine (sync.Once), evicts
  buckets inactive for 10min, Stop() closes stop channel
- gatewayRateLimiter.buckets: same pattern, started/stopped via
  APIGateway.Start/Stop lifecycle
- WebhookSender.deadLetter: 10k cap with oldest-first trim, hourly
  purge of entries older than 24h; Start/Stop lifecycle
- globalOAuthCache: sync.Once starts 5min cleanup goroutine on first
  getOrCreate; evicts entries whose token has expired
- InMemoryLock.locks: lastUsed tracked per entry, 10min cleanup tick
  evicts idle entries older than 30min; Start/Stop methods added

All cleanup goroutines include panic recovery and stop channels.
Tests added for each map verifying shrinkage after backdated entries.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: intel352

module/api_gateway.go

@@ -71,13 +71,15 @@ type gatewayRateLimiter struct {
 	buckets map[string]*tokenBucket
 	rpm     int
 	burst   int
+	stopCh  chan struct{}
 }
 
 func newGatewayRateLimiter(rpm, burst int) *gatewayRateLimiter {
 	return &gatewayRateLimiter{
 		buckets: make(map[string]*tokenBucket),
 		rpm:     rpm,
 		burst:   burst,
+		stopCh:  make(chan struct{}),
 	}
 }
 
@@ -94,6 +96,44 @@ func (rl *gatewayRateLimiter) allow(clientIP string) bool {
 	return bucket.allow()
 }
 
+// startCleanup launches a background goroutine that evicts stale buckets every 5 minutes.
+func (rl *gatewayRateLimiter) startCleanup() {
+	go func() {
+		defer func() { recover() }() //nolint:errcheck
+		ticker := time.NewTicker(5 * time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				rl.evictStaleBuckets(10 * time.Minute)
+			case <-rl.stopCh:
+				return
+			}
+		}
+	}()
+}
+
+// stop shuts down the background cleanup goroutine.
+func (rl *gatewayRateLimiter) stop() {
+	select {
+	case <-rl.stopCh:
+	default:
+		close(rl.stopCh)
+	}
+}
+
+// evictStaleBuckets removes buckets not accessed within ttl.
+func (rl *gatewayRateLimiter) evictStaleBuckets(ttl time.Duration) {
+	cutoff := time.Now().Add(-ttl)
+	rl.mu.Lock()
+	defer rl.mu.Unlock()
+	for k, b := range rl.buckets {
+		if b.lastRefill.Before(cutoff) {
+			delete(rl.buckets, k)
+		}
+	}
+}
+
 // APIGatewayOption is a functional option for configuring an APIGateway at construction time.
 type APIGatewayOption func(*APIGateway)
 
@@ -210,11 +250,27 @@ func (g *APIGateway) Name() string { return g.name }
 // Init initializes the module.
 func (g *APIGateway) Init(_ modular.Application) error { return nil }
 
-// Start is a no-op.
-func (g *APIGateway) Start(_ context.Context) error { return nil }
+// Start launches background cleanup goroutines for all rate limiters.
+func (g *APIGateway) Start(_ context.Context) error {
+	if g.instanceRateLimiter != nil {
+		g.instanceRateLimiter.startCleanup()
+	}
+	for _, rl := range g.rateLimiters {
+		rl.startCleanup()
+	}
+	return nil
+}
 
-// Stop is a no-op.
-func (g *APIGateway) Stop(_ context.Context) error { return nil }
+// Stop shuts down background cleanup goroutines for all rate limiters.
+func (g *APIGateway) Stop(_ context.Context) error {
+	if g.instanceRateLimiter != nil {
+		g.instanceRateLimiter.stop()
+	}
+	for _, rl := range g.rateLimiters {
+		rl.stop()
+	}
+	return nil
+}
 
 // ProvidesServices returns the services provided by this module.
 func (g *APIGateway) ProvidesServices() []modular.ServiceProvider {

module/api_gateway_test.go

@@ -1,6 +1,8 @@
 package module
 
 import (
+	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -497,3 +499,60 @@ func TestAWSAPIGateway_SyncRoutesRequiresAPIID(t *testing.T) {
 		t.Fatal("expected error when api_id is empty")
 	}
 }
+
+func TestGatewayRateLimiter_BucketEviction(t *testing.T) {
+	rl := newGatewayRateLimiter(60, 5)
+
+	// Fill with unique client IPs
+	for i := range 100 {
+		rl.allow(fmt.Sprintf("192.168.1.%d", i%256))
+	}
+
+	rl.mu.Lock()
+	count := len(rl.buckets)
+	rl.mu.Unlock()
+	if count == 0 {
+		t.Fatal("expected buckets to be populated")
+	}
+
+	// Backdate all buckets
+	rl.mu.Lock()
+	for _, b := range rl.buckets {
+		b.lastRefill = time.Now().Add(-20 * time.Minute)
+	}
+	rl.mu.Unlock()
+
+	rl.evictStaleBuckets(10 * time.Minute)
+
+	rl.mu.Lock()
+	count = len(rl.buckets)
+	rl.mu.Unlock()
+	if count != 0 {
+		t.Fatalf("expected 0 buckets after eviction, got %d", count)
+	}
+}
+
+func TestAPIGateway_StartStopRateLimiterCleanup(t *testing.T) {
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer backend.Close()
+
+	gw := NewAPIGateway("gw-lifecycle", WithRateLimit(&RateLimitConfig{
+		RequestsPerMinute: 60,
+		BurstSize:         10,
+	}))
+	if err := gw.SetRoutes([]GatewayRoute{
+		{PathPrefix: "/api", Backend: backend.URL, RateLimit: &RateLimitConfig{RequestsPerMinute: 30, BurstSize: 5}},
+	}); err != nil {
+		t.Fatalf("SetRoutes failed: %v", err)
+	}
+
+	ctx := context.Background()
+	if err := gw.Start(ctx); err != nil {
+		t.Fatalf("Start failed: %v", err)
+	}
+	if err := gw.Stop(ctx); err != nil {
+		t.Fatalf("Stop failed: %v", err)
+	}
+}

module/pipeline_step_http_call.go

@@ -23,16 +23,22 @@ import (
 // isolated entry.
 var globalOAuthCache = &oauthTokenCache{ //nolint:gochecknoglobals // intentional process-wide cache
 	entries: make(map[string]*oauthCacheEntry),
+	stopCh:  make(chan struct{}),
 }
 
 // oauthTokenCache is a registry of per-credential token cache entries.
 type oauthTokenCache struct {
-	mu      sync.RWMutex
-	entries map[string]*oauthCacheEntry
+	mu        sync.RWMutex
+	entries   map[string]*oauthCacheEntry
+	startOnce sync.Once
+	stopCh    chan struct{}
 }
 
 // getOrCreate returns the existing cache entry for key, or creates and stores a new one.
+// On first call it starts a background goroutine that evicts expired entries every 5 minutes.
 func (c *oauthTokenCache) getOrCreate(key string) *oauthCacheEntry {
+	c.startOnce.Do(func() { go c.cleanupLoop() })
+
 	c.mu.RLock()
 	entry, ok := c.entries[key]
 	c.mu.RUnlock()
@@ -49,6 +55,36 @@ func (c *oauthTokenCache) getOrCreate(key string) *oauthCacheEntry {
 	return entry
 }
 
+// cleanupLoop evicts expired entries from the cache every 5 minutes.
+func (c *oauthTokenCache) cleanupLoop() {
+	defer func() { recover() }() //nolint:errcheck
+	ticker := time.NewTicker(5 * time.Minute)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			c.evictExpired()
+		case <-c.stopCh:
+			return
+		}
+	}
+}
+
+// evictExpired removes entries whose tokens have expired.
+func (c *oauthTokenCache) evictExpired() {
+	now := time.Now()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for key, entry := range c.entries {
+		entry.mu.Lock()
+		expired := entry.accessToken == "" || now.After(entry.expiry)
+		entry.mu.Unlock()
+		if expired {
+			delete(c.entries, key)
+		}
+	}
+}
+
 // oauthCacheEntry holds a cached OAuth2 access token with expiry. A singleflight.Group is
 // embedded to ensure at most one concurrent token fetch per credential set.
 type oauthCacheEntry struct {

module/pipeline_step_http_call_test.go

@@ -1115,3 +1115,41 @@ func TestHTTPCallStep_ElapsedMS(t *testing.T) {
 		t.Errorf("expected elapsed_ms >= 0, got %d", elapsedMS)
 	}
 }
+
+func TestOAuthTokenCache_ExpiredEntryEviction(t *testing.T) {
+	cache := &oauthTokenCache{
+		entries: make(map[string]*oauthCacheEntry),
+		stopCh:  make(chan struct{}),
+	}
+
+	// Add entries: some expired, some valid
+	expiredEntry := &oauthCacheEntry{}
+	expiredEntry.set("expired-token", "", 1*time.Second)
+	expiredEntry.mu.Lock()
+	expiredEntry.expiry = time.Now().Add(-10 * time.Minute) // force expired
+	expiredEntry.mu.Unlock()
+
+	validEntry := &oauthCacheEntry{}
+	validEntry.set("valid-token", "", 1*time.Hour)
+
+	cache.mu.Lock()
+	cache.entries["expired-key"] = expiredEntry
+	cache.entries["valid-key"] = validEntry
+	cache.mu.Unlock()
+
+	cache.evictExpired()
+
+	cache.mu.RLock()
+	_, expiredStillPresent := cache.entries["expired-key"]
+	_, validStillPresent := cache.entries["valid-key"]
+	cache.mu.RUnlock()
+
+	if expiredStillPresent {
+		t.Error("expired entry should have been evicted")
+	}
+	if !validStillPresent {
+		t.Error("valid entry should not have been evicted")
+	}
+
+	close(cache.stopCh)
+}

module/pipeline_step_rate_limit.go

@@ -19,8 +19,10 @@ type RateLimitStep struct {
 	keyFrom           string // template for per-client key
 	tmpl              *TemplateEngine
 
-	mu      sync.Mutex
-	buckets map[string]*tokenBucket
+	mu        sync.Mutex
+	buckets   map[string]*tokenBucket
+	stopCh    chan struct{}
+	startOnce sync.Once
 }
 
 // tokenBucket implements a simple token bucket rate limiter.
@@ -97,16 +99,57 @@ func NewRateLimitStepFactory() StepFactory {
 			keyFrom:           keyFrom,
 			tmpl:              NewTemplateEngine(),
 			buckets:           make(map[string]*tokenBucket),
+			stopCh:            make(chan struct{}),
 		}, nil
 	}
 }
 
 // Name returns the step name.
 func (s *RateLimitStep) Name() string { return s.name }
 
+// Stop shuts down the background cleanup goroutine. Safe to call multiple times
+// only if the goroutine was started (i.e., Execute was called at least once).
+func (s *RateLimitStep) Stop() {
+	s.startOnce.Do(func() {}) // mark once as done so cleanup won't start if Stop races Execute
+	select {
+	case <-s.stopCh:
+	default:
+		close(s.stopCh)
+	}
+}
+
+// cleanupLoop removes stale token buckets every 5 minutes.
+func (s *RateLimitStep) cleanupLoop() {
+	defer func() { recover() }() //nolint:errcheck
+	ticker := time.NewTicker(5 * time.Minute)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			s.evictStaleBuckets(10 * time.Minute)
+		case <-s.stopCh:
+			return
+		}
+	}
+}
+
+// evictStaleBuckets removes buckets that have not been accessed within ttl.
+func (s *RateLimitStep) evictStaleBuckets(ttl time.Duration) {
+	cutoff := time.Now().Add(-ttl)
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for k, b := range s.buckets {
+		if b.lastRefill.Before(cutoff) {
+			delete(s.buckets, k)
+		}
+	}
+}
+
 // Execute checks rate limiting for the resolved key and either allows or
 // rejects the request.
 func (s *RateLimitStep) Execute(_ context.Context, pc *PipelineContext) (*StepResult, error) {
+	s.startOnce.Do(func() { go s.cleanupLoop() })
+
 	// Resolve the rate limit key from template
 	key := s.keyFrom
 	if s.tmpl != nil && key != "global" {