Make s3vectors have rate limit per namespace. #360
Conversation
WalkthroughReplaces a global rate limiter with a concurrency-safe per-namespace limiter utility and updates the S3Vector connector API and call sites to construct and use namespace-scoped limiters via an integer rateLimit parameter instead of a shared *rate.Limiter. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes
Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
📜 Recent review detailsConfiguration used: defaults Review profile: CHILL Plan: Pro 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (2)
🧰 Additional context used🧬 Code graph analysis (1)connectors/s3vector/conn.go (1)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
🔇 Additional comments (5)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
connectors/util/limiter.go (1)
38-57: Burst value may be negative when rate limit is infinite.When
v < 0ordefaultLimit < 0, the code setslimit = rate.Infbut still uses the negative value as the burst parameter inrate.NewLimiter(limit, v)(line 45) andrate.NewLimiter(limit, defaultLimit)(line 54). Whilerate.Limiterwithrate.Infallows unlimited tokens regardless of burst, a negative burst is semantically odd and may cause unexpected behavior if the limiter implementation changes.Consider using a sensible burst value (e.g.,
1or0) when the limit is infinite:for k, v := range namespaceToLimit { limit := rate.Limit(v) + burst := v if v < 0 { limit = rate.Inf + burst = 0 } - limiters[k] = rate.NewLimiter(limit, v) + limiters[k] = rate.NewLimiter(limit, burst) } limit := rate.Limit(defaultLimit) + burst := defaultLimit if defaultLimit < 0 { limit = rate.Inf + burst = 0 } return &namespaceLimiter{ limiters: limiters, defaultLimiterFactory: func() *rate.Limiter { - return rate.NewLimiter(limit, defaultLimit) + return rate.NewLimiter(limit, burst) }, }connectors/s3vector/conn.go (1)
221-256: Consider caching the namespace limiter to avoid repeated lookups.Within
WriteUpdates,limiter.Get(r.Msg.GetNamespace())is called in both the delete and put batch loops. While theGetmethod is thread-safe and efficient (read-lock for existing entries), caching the limiter before the loops would be slightly cleaner:if len(updates) > 0 { + nsLimiter := c.limiter.Get(r.Msg.GetNamespace()) eg, ctx := errgroup.WithContext(ctx) eg.SetLimit(c.maxParallelism) for batch := range slices.Chunk(toDelete, c.batchSize) { eg.Go(func() error { - if err := c.limiter.Get(r.Msg.GetNamespace()).WaitN(ctx, len(batch)); err != nil { + if err := nsLimiter.WaitN(ctx, len(batch)); err != nil { return fmt.Errorf("err in limiter: %w", err) } // ... }) } for batch := range slices.Chunk(vectors, c.batchSize) { eg.Go(func() error { - if err := c.limiter.Get(r.Msg.GetNamespace()).WaitN(ctx, len(batch)); err != nil { + if err := nsLimiter.WaitN(ctx, len(batch)); err != nil { return fmt.Errorf("err in limiter: %w", err) } // ... }) }
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
connectors/s3vector/conn.go(6 hunks)connectors/util/limiter.go(1 hunks)internal/app/options/connectorflags.go(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (3)
connectors/s3vector/conn.go (1)
connectors/util/limiter.go (2)
NamespaceLimiter(9-11)NewNamespaceLimiter(38-57)
connectors/util/limiter.go (1)
connectors/dynamodb/stream/helper.go (1)
Limiter(23-25)
internal/app/options/connectorflags.go (1)
connectors/s3vector/conn.go (1)
NewConn(261-275)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: build
🔇 Additional comments (7)
connectors/util/limiter.go (2)
9-11: LGTM!Clean interface definition that provides a clear contract for obtaining namespace-specific rate limiters.
19-36: LGTM!The double-checked locking pattern is correctly implemented for thread-safe lazy initialization of per-namespace limiters.
internal/app/options/connectorflags.go (2)
396-396: LGTM!The call to
NewConncorrectly passesrateLimitas an integer, aligning with the updated signature that now handles limiter construction internally.
380-388: Rate-limit and batch-size defaults align with AWS S3 Vectors service limits.The new rate-limit default of 2,500 vectors per second and batch-size default of 500 correspond to AWS S3 Vectors' documented maximum throughput (2,500 vectors per second per vector index) and maximum batch size per API request. These changes set the connector to recommended service maximums rather than conservative defaults, enabling optimal throughput for users without additional configuration. The previous values (500 rate-limit, 200 batch-size) were below the service's stated limits, so users hitting expected production loads would need manual tuning anyway. No action required.
connectors/s3vector/conn.go (3)
30-30: LGTM!The limiter field type correctly changed to
util.NamespaceLimiterto support per-namespace rate limiting.
89-103: LGTM!The rate limiter correctly retrieves a namespace-specific limiter before each batch operation. The closure safely captures
r.Msg.GetNamespace()since it's called at goroutine execution time (not captured by value at loop iteration).
261-275: LGTM!The constructor cleanly initializes the
NamespaceLimiterwithnilfor the namespace-specific map, meaning all namespaces will use the default rate limit. This provides uniform per-namespace limiting as intended by the PR.
adiom-mark
force-pushed
the
s3vectorsratelimit
branch
from
b844cbb to
4f278ee
Compare
Summary by CodeRabbit
New Features
Refactor
✏️ Tip: You can customize this high-level summary in your review settings.