1/**
2 * Coordinates rate limits through shared persistent storage.
3 *
4 * The `RateLimiter` service consumes tokens for string keys using fixed-window
5 * counters or token-bucket state. It can protect external APIs, enforce quotas,
6 * or throttle workers across fibers and processes that share the same store.
7 * This module includes helpers that fail when a limit is exceeded, return the
8 * delay needed before continuing, or wrap an effect so it waits automatically.
9 * It also defines the store service and in-memory or Redis-backed store layers.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Config from "../Config.ts"
15import * as Context from "../Context.ts"
16import * as Duration from "../Duration.ts"
17import * as Effect from "../Effect.ts"
18import { flow, identity } from "../Function.ts"
19import * as Layer from "../Layer.ts"
20import * as Schema from "../Schema.ts"
21import * as Redis from "./Redis.ts"
22
23/**
24 * Runtime type identifier for `RateLimiter` values.
25 *
26 * @stability unstable
27 * @category type IDs
28 * @since 4.0.0
29 */
30export const TypeId: TypeId = "~effect/persistence/RateLimiter"
31
32/**
33 * Type-level identifier used to brand `RateLimiter` values.
34 *
35 * @stability unstable
36 * @category type IDs
37 * @since 4.0.0
38 */
39export type TypeId = "~effect/persistence/RateLimiter"
40
41/**
42 * Service for consuming rate-limit tokens for a key using fixed-window or
43 * token-bucket algorithms.
44 *
45 * @stability unstable
46 * @category models
47 * @since 4.0.0
48 */
49export interface RateLimiter {
50 readonly [TypeId]: TypeId
51
52 readonly consume: (options: {
53 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
54 readonly onExceeded?: "delay" | "fail" | undefined
55 readonly window: Duration.Input
56 readonly limit: number
57 readonly key: string
58 readonly tokens?: number | undefined
59 }) => Effect.Effect<ConsumeResult, RateLimiterError>
60
61 readonly adaptiveConsume: (options: AdaptiveConsumeOptions) => Effect.Effect<AdaptiveConsumeResult, RateLimiterError>
62
63 readonly adaptiveFeedback: (options: AdaptiveFeedbackOptions) => Effect.Effect<void, RateLimiterError>
64}
65
66/**
67 * Service tag for persistent token-consumption services.
68 *
69 * **When to use**
70 *
71 * Use to access or provide rate-limit checks backed by fixed-window counters or
72 * token-bucket state.
73 *
74 * @stability unstable
75 * @category services
76 * @since 4.0.0
77 */
78export const RateLimiter: Context.Service<RateLimiter, RateLimiter> = Context.Service<RateLimiter>(TypeId)
79
80/**
81 * Creates a `RateLimiter` from the current `RateLimiterStore`.
82 *
83 * **Details**
84 *
85 * The limiter supports fixed-window and token-bucket algorithms and either
86 * fails or returns a delay when a limit is exceeded.
87 *
88 * @stability unstable
89 * @category constructors
90 * @since 4.0.0
91 */
92export const make: Effect.Effect<
93 RateLimiter,
94 never,
95 RateLimiterStore
96> = Effect.gen(function*() {
97 const store = yield* RateLimiterStore
98
99 return identity<RateLimiter>({
100 [TypeId]: TypeId,
101 adaptiveConsume: store.adaptiveConsume,
102 adaptiveFeedback: store.adaptiveFeedback,
103 consume(options) {
104 const tokens = options.tokens ?? 1
105 const onExceeded = options.onExceeded ?? "fail"
106 const algorithm = options.algorithm ?? "fixed-window"
107 const window = Duration.max(Duration.fromInputUnsafe(options.window), Duration.millis(1))
108 const windowMillis = Duration.toMillis(window)
109 const refillRate = Duration.divideUnsafe(window, options.limit)
110 const refillRateMillis = Duration.toMillis(refillRate)
111
112 if (tokens > options.limit) {
113 return onExceeded === "fail"
114 ? Effect.fail(
115 new RateLimiterError({
116 reason: new RateLimitExceeded({
117 key: options.key,
118 retryAfter: window,
119 limit: options.limit,
120 remaining: 0
121 })
122 })
123 )
124 : Effect.succeed<ConsumeResult>({
125 delay: window,
126 limit: options.limit,
127 remaining: 0,
128 resetAfter: window
129 })
130 }
131
132 if (algorithm === "fixed-window") {
133 return Effect.flatMap(
134 store.fixedWindow({
135 key: options.key,
136 tokens,
137 refillRate,
138 limit: onExceeded === "fail" ? options.limit : undefined
139 }),
140 ([count, ttl]) => {
141 if (onExceeded === "fail") {
142 const remaining = options.limit - count
143 if (remaining < 0) {
144 return Effect.fail(
145 new RateLimiterError({
146 reason: new RateLimitExceeded({
147 key: options.key,
148 retryAfter: Duration.millis(ttl),
149 limit: options.limit,
150 remaining: 0
151 })
152 })
153 )
154 }
155 return Effect.succeed<ConsumeResult>({
156 delay: Duration.zero,
157 limit: options.limit,
158 remaining,
159 resetAfter: Duration.millis(ttl)
160 })
161 }
162 const ttlTotal = count * refillRateMillis
163 const elapsed = ttlTotal - ttl
164 const windowNumber = Math.floor((count - 1) / options.limit)
165 const remaining = (windowNumber * windowMillis) - elapsed
166 const delay = remaining <= 0 ? Duration.zero : Duration.millis(remaining)
167 return Effect.succeed<ConsumeResult>({
168 delay,
169 limit: options.limit,
170 remaining: options.limit - count,
171 resetAfter: Duration.millis(ttl)
172 })
173 }
174 )
175 }
176
177 return Effect.flatMap(
178 store.tokenBucket({
179 key: options.key,
180 tokens,
181 limit: options.limit,
182 refillRate,
183 allowOverflow: onExceeded === "delay"
184 }),
185 ([remaining, elapsedMillis]) => {
186 const delay = Duration.millis(Math.max(0, Math.ceil(-remaining) * refillRateMillis - elapsedMillis))
187 const resetAfter = Duration.millis(
188 Math.max(0, Math.ceil(options.limit - remaining) * refillRateMillis - elapsedMillis)
189 )
190 if (onExceeded === "fail" && remaining < 0) {
191 return Effect.fail(
192 new RateLimiterError({
193 reason: new RateLimitExceeded({
194 key: options.key,
195 retryAfter: delay,
196 limit: options.limit,
197 remaining: 0
198 })
199 })
200 )
201 }
202 return Effect.succeed<ConsumeResult>({
203 delay,
204 limit: options.limit,
205 remaining,
206 resetAfter
207 })
208 }
209 )
210 }
211 })
212})
213
214/**
215 * Provides `RateLimiter` using the current `RateLimiterStore`.
216 *
217 * @stability unstable
218 * @category layers
219 * @since 4.0.0
220 */
221export const layer: Layer.Layer<
222 RateLimiter,
223 never,
224 RateLimiterStore
225> = Layer.effect(RateLimiter, make)
226
227/**
228 * Accesses a function that applies rate limiting to an effect.
229 *
230 * **Example** (Applying rate limits to effects)
231 *
232 * ```ts import.meta.vitest
233 * import { Effect, Layer } from "effect"
234 * import { RateLimiter } from "effect/persistence"
235 *
236 * const messages: Array<string> = []
237 * const program = Effect.gen(function*() {
238 * // Access the `withLimiter` function from the RateLimiter module
239 * const withLimiter = yield* RateLimiter.makeWithRateLimiter
240 *
241 * // Apply a rate limiter to an effect
242 * yield* Effect.sync(() => messages.push("Making a request with rate limiting")).pipe(
243 * withLimiter({
244 * key: "some-key",
245 * limit: 10,
246 * onExceeded: "delay",
247 * window: "5 seconds",
248 * algorithm: "fixed-window"
249 * })
250 * )
251 * }).pipe(
252 * Effect.provide(RateLimiter.layer.pipe(Layer.provide(RateLimiter.layerStoreMemory)))
253 * )
254 *
255 * await Effect.runPromise(program)
256 * messages // => ["Making a request with rate limiting"]
257 * ```
258 *
259 * @stability unstable
260 * @category accessors
261 * @since 4.0.0
262 */
263export const makeWithRateLimiter: Effect.Effect<
264 ((options: {
265 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
266 readonly onExceeded?: "delay" | "fail" | undefined
267 readonly window: Duration.Input
268 readonly limit: number
269 readonly key: string
270 readonly tokens?: number | undefined
271 }) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E | RateLimiterError, R>),
272 never,
273 RateLimiter
274> = RateLimiter.use((limiter) =>
275 Effect.succeed((options) => (effect) =>
276 Effect.flatMap(limiter.consume(options), ({ delay }) => {
277 if (Duration.isZero(delay)) return effect
278 return Effect.delay(effect, delay)
279 })
280 )
281)
282
283/**
284 * Sleeps when the rate limit is exceeded.
285 *
286 * **Example** (Sleeping until rate limit permits)
287 *
288 * ```ts import.meta.vitest
289 * import { Effect, Layer } from "effect"
290 * import { RateLimiter } from "effect/persistence"
291 *
292 * const program = Effect.gen(function*() {
293 * const limiter = yield* RateLimiter.RateLimiter
294 * const partiallyApplied = RateLimiter.sleep(limiter)
295 * const partial = yield* partiallyApplied({
296 * key: "partial",
297 * limit: 10,
298 * window: "5 seconds",
299 * algorithm: "fixed-window"
300 * })
301 * const direct = yield* RateLimiter.sleep(limiter, {
302 * key: "direct",
303 * limit: 10,
304 * window: "5 seconds",
305 * algorithm: "fixed-window"
306 * })
307 * return [partial.remaining, direct.remaining]
308 * }).pipe(
309 * Effect.provide(RateLimiter.layer.pipe(Layer.provide(RateLimiter.layerStoreMemory)))
310 * )
311 *
312 * await Effect.runPromise(program) // => [9, 9]
313 * ```
314 *
315 * @stability unstable
316 * @category accessors
317 * @since 4.0.0
318 */
319export function sleep(self: RateLimiter): (options: {
320 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
321 readonly window: Duration.Input
322 readonly limit: number
323 readonly key: string
324 readonly tokens?: number | undefined
325}) => Effect.Effect<ConsumeResult, RateLimiterError>
326/**
327 * @stability unstable
328 */
329export function sleep(self: RateLimiter, options: {
330 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
331 readonly window: Duration.Input
332 readonly limit: number
333 readonly key: string
334 readonly tokens?: number | undefined
335}): Effect.Effect<ConsumeResult, RateLimiterError>
336/**
337 * @stability unstable
338 */
339export function sleep(self: RateLimiter, options?: {
340 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
341 readonly window: Duration.Input
342 readonly limit: number
343 readonly key: string
344 readonly tokens?: number | undefined
345}):
346 | Effect.Effect<ConsumeResult, RateLimiterError>
347 | ((options: {
348 readonly algorithm?: "fixed-window" | "token-bucket" | undefined
349 readonly window: Duration.Input
350 readonly limit: number
351 readonly key: string
352 readonly tokens?: number | undefined
353 }) => Effect.Effect<ConsumeResult, RateLimiterError>)
354{
355 if (options === undefined) {
356 return (options) => sleep(self, options)
357 }
358 return Effect.flatMap(
359 self.consume({
360 ...options,
361 onExceeded: "delay"
362 }),
363 (result) => {
364 if (Duration.isZero(result.delay)) return Effect.succeed(result)
365 return Effect.as(Effect.sleep(result.delay), result)
366 }
367 )
368}
369
370/**
371 * Runtime type identifier for `RateLimiterError`.
372 *
373 * @stability unstable
374 * @category type IDs
375 * @since 4.0.0
376 */
377export const ErrorTypeId: ErrorTypeId = "~effect/persistence/RateLimiter/RateLimiterError"
378
379/**
380 * Type-level identifier used to brand `RateLimiterError` values.
381 *
382 * @stability unstable
383 * @category type IDs
384 * @since 4.0.0
385 */
386export type ErrorTypeId = "~effect/persistence/RateLimiter/RateLimiterError"
387
388/**
389 * Error reason for a rate-limit check that exceeded the configured limit.
390 *
391 * **Details**
392 *
393 * Includes the affected key, limit, remaining token count, and retry delay.
394 *
395 * @stability unstable
396 * @category errors
397 * @since 4.0.0
398 */
399export class RateLimitExceeded extends Schema.Error<RateLimitExceeded>(
400 "effect/persistence/RateLimiter/RateLimitExceeded"
401)({
402 _tag: Schema.tag("RateLimitExceeded"),
403 retryAfter: Schema.DurationFromMillis,
404 key: Schema.String,
405 limit: Schema.Finite,
406 remaining: Schema.Finite
407}) {
408 /**
409 * Public message used when the rate limiter rejects a request.
410 *
411 * @stability unstable
412 * @since 4.0.0
413 */
414 override get message(): string {
415 return `Rate limit exceeded`
416 }
417}
418
419/**
420 * Error reason for failures in the backing `RateLimiterStore`.
421 *
422 * @stability unstable
423 * @category errors
424 * @since 4.0.0
425 */
426export class RateLimitStoreError extends Schema.Error<RateLimitStoreError>(
427 "effect/persistence/RateLimiter/RateLimitStoreError"
428)({
429 _tag: Schema.tag("RateLimitStoreError"),
430 message: Schema.String,
431 cause: Schema.optional(Schema.Defect())
432}) {}
433
434/**
435 * Union of reasons carried by `RateLimiterError`.
436 *
437 * @stability unstable
438 * @category errors
439 * @since 4.0.0
440 */
441export type RateLimiterErrorReason = RateLimitExceeded | RateLimitStoreError
442
443/**
444 * Schema for all reasons that can be carried by `RateLimiterError`.
445 *
446 * @stability unstable
447 * @category errors
448 * @since 4.0.0
449 */
450export const RateLimiterErrorReason: Schema.Union<[
451 typeof RateLimitExceeded,
452 typeof RateLimitStoreError
453]> = Schema.Union([RateLimitExceeded, RateLimitStoreError])
454
455/**
456 * Error raised by rate limiter operations, wrapping a concrete failure
457 * `reason`.
458 *
459 * @stability unstable
460 * @category errors
461 * @since 4.0.0
462 */
463export class RateLimiterError extends Schema.Error<RateLimiterError>(ErrorTypeId)({
464 _tag: Schema.tag("RateLimiterError"),
465 reason: RateLimiterErrorReason
466}) {
467 // @effect-diagnostics-next-line overriddenSchemaConstructor:off
468 constructor(props: {
469 readonly reason: RateLimiterErrorReason
470 }) {
471 if ("cause" in props.reason) {
472 super({
473 ...props,
474 cause: props.reason.cause
475 } as any)
476 } else {
477 super(props)
478 }
479 }
480
481 /**
482 * Marks this value as a rate limiter error for runtime guards.
483 *
484 * @stability unstable
485 * @since 4.0.0
486 */
487 readonly [ErrorTypeId]: ErrorTypeId = ErrorTypeId
488
489 override get message(): string {
490 return this.reason.message
491 }
492}
493
494/**
495 * Metadata returned after consuming tokens from a rate limiter.
496 *
497 * @stability unstable
498 * @category models
499 * @since 4.0.0
500 */
501export interface ConsumeResult {
502 /**
503 * The amount of delay to wait before making the next request, when the rate
504 * limiter is using the "delay" `onExceeded` strategy. It will be
505 * Duration.zero if the request is allowed immediately.
506 */
507 readonly delay: Duration.Duration
508
509 /**
510 * The maximum number of requests allowed in the current window.
511 */
512 readonly limit: number
513
514 /**
515 * The number of remaining requests in the current window.
516 */
517 readonly remaining: number
518
519 /**
520 * The time until the rate limit fully resets.
521 *
522 * **Details**
523 *
524 * For token buckets, accounts for elapsed refill time and reserved debt,
525 * assuming no further consumption.
526 */
527 readonly resetAfter: Duration.Duration
528}
529
530/**
531 * Phase of adaptive rate limiting driven by server feedback.
532 *
533 * @stability unstable
534 * @category models
535 * @since 4.0.0
536 */
537export type AdaptivePhase = "inactive" | "cooldown" | "learning" | "learned"
538
539/**
540 * Options for consuming tokens from the adaptive rate limiter store.
541 *
542 * @stability unstable
543 * @category options
544 * @since 4.0.0
545 */
546export interface AdaptiveConsumeOptions {
547 /**
548 * The rate-limit key.
549 */
550 readonly key: string
551
552 /**
553 * The number of tokens to consume.
554 */
555 readonly tokens: number
556
557 /**
558 * The fallback limit configured for the regular rate limiter.
559 */
560 readonly fallbackLimit: number
561
562 /**
563 * The fallback window configured for the regular rate limiter.
564 */
565 readonly fallbackWindow: Duration.Duration
566}
567
568/**
569 * Metadata returned after consuming tokens from the adaptive rate limiter store.
570 *
571 * @stability unstable
572 * @category models
573 * @since 4.0.0
574 */
575export interface AdaptiveConsumeResult {
576 /**
577 * The amount of delay to wait before making the request.
578 */
579 readonly delay: Duration.Duration
580
581 /**
582 * The adaptive state epoch used to correlate later response feedback.
583 */
584 readonly epoch: number
585
586 /**
587 * The adaptive phase observed by this consume operation.
588 */
589 readonly phase: AdaptivePhase
590}
591
592/**
593 * Options for reporting response feedback to the adaptive rate limiter store.
594 *
595 * @stability unstable
596 * @category options
597 * @since 4.0.0
598 */
599export interface AdaptiveFeedbackOptions {
600 /**
601 * The rate-limit key.
602 */
603 readonly key: string
604
605 /**
606 * The adaptive state epoch returned by `adaptiveConsume`.
607 */
608 readonly epoch: number
609
610 /**
611 * The number of tokens consumed by the request.
612 */
613 readonly tokens: number
614
615 /**
616 * The HTTP response status code.
617 */
618 readonly status: number
619
620 /**
621 * The parsed `Retry-After` delay, when present.
622 */
623 readonly retryAfter: Duration.Duration | undefined
624}
625
626/**
627 * Defines the low-level backing store for rate-limit state.
628 *
629 * **When to use**
630 *
631 * Use to provide the shared counter storage and adaptive feedback state used by
632 * persistent rate-limit checks.
633 *
634 * @stability unstable
635 * @category services
636 * @since 4.0.0
637 */
638export class RateLimiterStore extends Context.Service<
639 RateLimiterStore,
640 {
641 /**
642 * Returns the token count *after* taking the specified `tokens` and time to
643 * live for the `key`.
644 *
645 * If `limit` is provided, the number of taken tokens will be capped at the
646 * limit.
647 *
648 * In the case the limit is exceeded, the returned count will be greater
649 * than the limit, but the TTL will not be updated.
650 */
651 readonly fixedWindow: (options: {
652 readonly key: string
653 readonly tokens: number
654 readonly refillRate: Duration.Duration
655 readonly limit: number | undefined
656 }) => Effect.Effect<readonly [count: number, ttl: number], RateLimiterError>
657
658 /**
659 * Refills the bucket for `key`, attempts to consume `tokens`, and returns
660 * `[remaining, elapsedMillis]` from that single atomic operation.
661 *
662 * `remaining` is the token count after subtracting `tokens`. Fractional counts
663 * must retain their numeric precision. A negative count is only persisted
664 * when `allowOverflow` is true.
665 *
666 * `elapsedMillis` is the time since the current refill interval started, in
667 * milliseconds (fractions preserved), always at least `0` and less than
668 * `Duration.toMillis(refillRate)`. It is `0` when the bucket is at capacity
669 * after refilling and before consuming. Otherwise the boundary advances by
670 * whole refill intervals and the interval never restarts.
671 */
672 readonly tokenBucket: (options: {
673 readonly key: string
674 readonly tokens: number
675 readonly limit: number
676 readonly refillRate: Duration.Duration
677 readonly allowOverflow: boolean
678 }) => Effect.Effect<readonly [remaining: number, elapsedMillis: number], RateLimiterError>
679
680 /**
681 * Consumes tokens from the adaptive rate-limit state for the `key`.
682 *
683 * When the store has no adaptive state for the `key`, implementations
684 * should return a zero delay with the inactive phase.
685 */
686 readonly adaptiveConsume: (
687 options: AdaptiveConsumeOptions
688 ) => Effect.Effect<AdaptiveConsumeResult, RateLimiterError>
689
690 /**
691 * Records response feedback for the adaptive rate-limit state.
692 */
693 readonly adaptiveFeedback: (options: AdaptiveFeedbackOptions) => Effect.Effect<void, RateLimiterError>
694 }
695>()("effect/persistence/RateLimiter/RateLimiterStore") {}
696
697const adaptiveStateTtlGraceMillis = 60_000
698const adaptiveStateMaxWindowMillis = 60 * 60 * 1_000
699
700const clampAdaptiveDurationMillis = (millis: number): number => {
701 if (Number.isNaN(millis) || millis <= 0) return 1
702 return Math.min(millis, adaptiveStateMaxWindowMillis)
703}
704
705interface AdaptiveState {
706 phase: AdaptivePhase
707 epoch: number
708 cooldownUntil: number
709 learningStartedAt: number
710 observedTokens: number
711 learnedLimit: number
712 learnedWindowMillis: number
713 expiresAt: number
714}
715
716/**
717 * Provides a process-local in-memory `RateLimiterStore`.
718 *
719 * @stability unstable
720 * @category layers
721 * @since 4.0.0
722 */
723export const layerStoreMemory: Layer.Layer<
724 RateLimiterStore
725> = Layer.sync(RateLimiterStore, () => {
726 const fixedCounters = new Map<string, { count: number; expiresAt: number }>()
727 const tokenBuckets = new Map<string, { tokens: number; lastRefill: number }>()
728 const adaptiveStates = new Map<string, AdaptiveState>()
729
730 const getAdaptiveState = (key: string, now: number): AdaptiveState | undefined => {
731 const state = adaptiveStates.get(key)
732 if (!state) return undefined
733 if (state.expiresAt <= now) {
734 adaptiveStates.delete(key)
735 return undefined
736 }
737 return state
738 }
739
740 const cooldownExpiresAt = (cooldownUntil: number): number => cooldownUntil + adaptiveStateTtlGraceMillis
741
742 const learningExpiresAt = (now: number, fallbackWindow: Duration.Duration): number =>
743 now + Duration.toMillis(fallbackWindow) + adaptiveStateTtlGraceMillis
744
745 const learnedExpiresAt = (now: number, learnedWindowMillis: number): number =>
746 now + learnedWindowMillis + adaptiveStateTtlGraceMillis
747
748 return RateLimiterStore.of({
749 fixedWindow: (options) =>
750 Effect.clockWith((clock) =>
751 Effect.sync(() => {
752 const refillRateMillis = Duration.toMillis(options.refillRate)
753 const now = clock.currentTimeMillisUnsafe()
754 let counter = fixedCounters.get(options.key)
755 if (!counter || counter.expiresAt <= now) {
756 counter = { count: 0, expiresAt: now }
757 fixedCounters.set(options.key, counter)
758 }
759 if (options.limit && counter.count + options.tokens > options.limit) {
760 return [counter.count + options.tokens, counter.expiresAt - now] as const
761 }
762 counter.count += options.tokens
763 counter.expiresAt += refillRateMillis * options.tokens
764 return [counter.count, counter.expiresAt - now] as const
765 })
766 ),
767 tokenBucket: (options) =>
768 Effect.clockWith((clock) =>
769 Effect.sync(() => {
770 const refillRateMillis = Duration.toMillis(options.refillRate)
771 const now = clock.currentTimeMillisUnsafe()
772 let bucket = tokenBuckets.get(options.key)
773 if (!bucket) {
774 bucket = { tokens: options.limit, lastRefill: now }
775 tokenBuckets.set(options.key, bucket)
776 }
777 const tokensToAdd = Math.floor((now - bucket.lastRefill) / refillRateMillis)
778 if (tokensToAdd > 0) {
779 bucket.tokens = Math.min(options.limit, bucket.tokens + tokensToAdd)
780 bucket.lastRefill += tokensToAdd * refillRateMillis
781 }
782 if (bucket.tokens >= options.limit) {
783 bucket.lastRefill = now
784 }
785
786 const newTokenCount = bucket.tokens - options.tokens
787 if (options.allowOverflow || newTokenCount >= 0) {
788 bucket.tokens = newTokenCount
789 }
790 return [newTokenCount, Math.max(0, now - bucket.lastRefill)] as const
791 })
792 ),
793 adaptiveConsume: (options) =>
794 Effect.clockWith((clock) =>
795 Effect.sync(() => {
796 const now = clock.currentTimeMillisUnsafe()
797 const state = getAdaptiveState(options.key, now)
798 if (!state) {
799 return {
800 delay: Duration.zero,
801 epoch: 0,
802 phase: "inactive"
803 }
804 }
805
806 if (state.phase === "cooldown") {
807 if (state.cooldownUntil > now) {
808 return {
809 delay: Duration.millis(state.cooldownUntil - now),
810 epoch: state.epoch,
811 phase: "cooldown"
812 }
813 }
814
815 state.phase = "learning"
816 state.epoch += 1
817 state.learningStartedAt = now
818 state.observedTokens = options.tokens
819 state.expiresAt = learningExpiresAt(now, options.fallbackWindow)
820 return {
821 delay: Duration.zero,
822 epoch: state.epoch,
823 phase: "learning"
824 }
825 }
826
827 if (state.phase === "learning") {
828 state.observedTokens += options.tokens
829 return {
830 delay: Duration.zero,
831 epoch: state.epoch,
832 phase: state.phase
833 }
834 }
835
836 if (state.phase === "learned") {
837 const refillRateMillis = state.learnedWindowMillis / state.learnedLimit
838 if (state.cooldownUntil <= now) {
839 state.observedTokens = 0
840 state.cooldownUntil = now
841 }
842 state.observedTokens += options.tokens
843 state.cooldownUntil += refillRateMillis * options.tokens
844
845 const ttl = state.cooldownUntil - now
846 const ttlTotal = state.observedTokens * refillRateMillis
847 const elapsed = ttlTotal - ttl
848 const windowNumber = Math.floor((state.observedTokens - 1) / state.learnedLimit)
849 const remaining = (windowNumber * state.learnedWindowMillis) - elapsed
850
851 return {
852 delay: remaining <= 0 ? Duration.zero : Duration.millis(remaining),
853 epoch: state.epoch,
854 phase: state.phase
855 }
856 }
857
858 return {
859 delay: Duration.zero,
860 epoch: state.epoch,
861 phase: state.phase
862 }
863 })
864 ),
865 adaptiveFeedback: (options) =>
866 Effect.clockWith((clock) =>
867 Effect.sync(() => {
868 if (options.status !== 429 || options.retryAfter === undefined) return
869
870 const retryAfterMillis = clampAdaptiveDurationMillis(Duration.toMillis(options.retryAfter))
871
872 const now = clock.currentTimeMillisUnsafe()
873 const cooldownUntil = now + retryAfterMillis
874 const state = getAdaptiveState(options.key, now)
875 if (!state) {
876 if (options.epoch !== 0) return
877 adaptiveStates.set(options.key, {
878 phase: "cooldown",
879 epoch: 0,
880 cooldownUntil,
881 learningStartedAt: 0,
882 observedTokens: 0,
883 learnedLimit: 0,
884 learnedWindowMillis: 0,
885 expiresAt: cooldownExpiresAt(cooldownUntil)
886 })
887 return
888 }
889
890 if (state.epoch !== options.epoch) return
891
892 if (state.phase === "cooldown") {
893 state.cooldownUntil = Math.max(state.cooldownUntil, cooldownUntil)
894 state.expiresAt = cooldownExpiresAt(state.cooldownUntil)
895 return
896 }
897
898 if (state.phase === "learning") {
899 const acceptedTokens = state.observedTokens - options.tokens
900 if (acceptedTokens <= 0) {
901 state.phase = "cooldown"
902 state.cooldownUntil = cooldownUntil
903 state.learningStartedAt = 0
904 state.observedTokens = 0
905 state.learnedLimit = 0
906 state.learnedWindowMillis = 0
907 state.expiresAt = cooldownExpiresAt(cooldownUntil)
908 return
909 }
910
911 const learnedWindowMillis = clampAdaptiveDurationMillis((now - state.learningStartedAt) + retryAfterMillis)
912 state.phase = "learned"
913 state.epoch += 1
914 state.cooldownUntil = state.learningStartedAt + learnedWindowMillis
915 state.observedTokens = acceptedTokens
916 state.learnedLimit = acceptedTokens
917 state.learnedWindowMillis = learnedWindowMillis
918 state.expiresAt = learnedExpiresAt(now, learnedWindowMillis)
919 return
920 }
921
922 if (state.phase === "learned") {
923 state.phase = "cooldown"
924 state.cooldownUntil = cooldownUntil
925 state.learningStartedAt = 0
926 state.observedTokens = 0
927 state.learnedLimit = 0
928 state.learnedWindowMillis = 0
929 state.expiresAt = cooldownExpiresAt(cooldownUntil)
930 }
931 })
932 )
933 })
934})
935
936/**
937 * Creates a Redis-backed `RateLimiterStore` using Lua scripts and the
938 * configured key prefix.
939 *
940 * @stability unstable
941 * @category constructors
942 * @since 4.0.0
943 */
944export const makeStoreRedis = Effect.fnUntraced(function*(
945 options?: {
946 readonly prefix?: string | undefined
947 }
948) {
949 const prefix = options?.prefix ?? "ratelimiter:"
950 const redis = yield* Redis.Redis
951
952 const fixedWindow = redis.eval(fixedWindowScript)
953 const tokenBucket = redis.eval(tokenBucketScript)
954 const adaptiveConsume = redis.eval(adaptiveConsumeScript)
955 const adaptiveFeedback = redis.eval(adaptiveFeedbackScript)
956
957 return RateLimiterStore.of({
958 fixedWindow(options) {
959 const key = `${prefix}${options.key}`
960 const refillMillis = Duration.toMillis(options.refillRate)
961 return Effect.mapError(
962 fixedWindow(key, options.tokens, refillMillis, options.limit),
963 (cause) =>
964 new RateLimiterError({
965 reason: new RateLimitStoreError({
966 message: `Failed to execute fixedWindow rate limiting command`,
967 cause: cause.cause
968 })
969 })
970 )
971 },
972 tokenBucket(options) {
973 const key = `${prefix}${options.key}`
974 const lastRefillKey = `${key}:refill`
975 const refillMillis = Duration.toMillis(options.refillRate)
976 return Effect.clockWith((clock) =>
977 Effect.mapError(
978 Effect.map(
979 tokenBucket(
980 key,
981 lastRefillKey,
982 options.tokens,
983 refillMillis,
984 options.limit,
985 clock.currentTimeMillisUnsafe(),
986 options.allowOverflow ? 1 : 0
987 ),
988 ([remaining, elapsedMillis]) => [Number(remaining), Number(elapsedMillis)] as const
989 ),
990 (cause) =>
991 new RateLimiterError({
992 reason: new RateLimitStoreError({
993 message: `Failed to execute tokenBucket rate limiting command`,
994 cause
995 })
996 })
997 )
998 )
999 },
1000 adaptiveConsume(options) {
1001 const key = `${prefix}${options.key}:adaptive`
1002 return Effect.map(
1003 Effect.mapError(
1004 adaptiveConsume(
1005 key,
1006 options.tokens,
1007 Duration.toMillis(options.fallbackWindow),
1008 adaptiveStateTtlGraceMillis
1009 ),
1010 (cause) =>
1011 new RateLimiterError({
1012 reason: new RateLimitStoreError({
1013 message: `Failed to execute adaptiveConsume rate limiting command`,
1014 cause: cause.cause
1015 })
1016 })
1017 ),
1018 ([delayMillis, epoch, phase]) => ({
1019 delay: delayMillis <= 0 ? Duration.zero : Duration.millis(delayMillis),
1020 epoch,
1021 phase
1022 })
1023 )
1024 },
1025 adaptiveFeedback(options) {
1026 if (options.status !== 429 || options.retryAfter === undefined) return Effect.void
1027 const retryAfterMillis = clampAdaptiveDurationMillis(Duration.toMillis(options.retryAfter))
1028 const key = `${prefix}${options.key}:adaptive`
1029 return Effect.asVoid(
1030 Effect.mapError(
1031 adaptiveFeedback(
1032 key,
1033 options.epoch,
1034 options.tokens,
1035 retryAfterMillis,
1036 adaptiveStateTtlGraceMillis,
1037 adaptiveStateMaxWindowMillis
1038 ),
1039 (cause) =>
1040 new RateLimiterError({
1041 reason: new RateLimitStoreError({
1042 message: `Failed to execute adaptiveFeedback rate limiting command`,
1043 cause: cause.cause
1044 })
1045 })
1046 )
1047 )
1048 }
1049 })
1050})
1051
1052const fixedWindowScript = Redis.script(
1053 (key: string, tokens: number, refillMillis: number, limit?: number) => [key, tokens, refillMillis, limit],
1054 {
1055 numberOfKeys: 1,
1056 lua: `
1057local key = KEYS[1]
1058local tokens = tonumber(ARGV[1])
1059local refillms = tonumber(ARGV[2])
1060local limit = tonumber(ARGV[3])
1061local current = tonumber(redis.call("GET", key))
1062
1063if not current then
1064 local nextpttl = math.max(1, math.ceil(refillms * tokens))
1065 redis.call("SET", key, tokens, "PX", nextpttl)
1066 return { tokens, nextpttl }
1067end
1068
1069local currentpttl = tonumber(redis.call("PTTL", key) or "0")
1070local next = current + tokens
1071if limit and next > limit then
1072 return { next, currentpttl }
1073end
1074
1075local nextpttl = math.max(1, math.ceil(currentpttl + (refillms * tokens)))
1076redis.call("SET", key, next, "PX", nextpttl)
1077return { next, nextpttl }
1078`
1079 }
1080).withReturnType<readonly [currentTokens: number, nextPttl: number]>()
1081
1082const tokenBucketScript = Redis.script(
1083 (
1084 key: string,
1085 lastRefillKey: string,
1086 tokens: number,
1087 refillMillis: number,
1088 limit: number,
1089 now: number,
1090 overflow: 0 | 1
1091 ) => [key, lastRefillKey, tokens, refillMillis, limit, now, overflow],
1092 {
1093 numberOfKeys: 2,
1094 lua: `
1095local key = KEYS[1]
1096local last_refill_key = KEYS[2]
1097local tokens = tonumber(ARGV[1])
1098local refill_ms = tonumber(ARGV[2])
1099local limit = tonumber(ARGV[3])
1100local now = tonumber(ARGV[4])
1101local overflow = ARGV[5] == "1"
1102local current = tonumber(redis.call("GET", key))
1103local last_refill = tonumber(redis.call("GET", last_refill_key))
1104
1105if not current then current = limit end
1106if not last_refill then last_refill = now end
1107
1108local elapsed = now - last_refill
1109local refill_amount = math.floor(elapsed / refill_ms)
1110if refill_amount > 0 then
1111 current = math.min(current + refill_amount, limit)
1112 last_refill = last_refill + (refill_amount * refill_ms)
1113end
1114if current >= limit then
1115 last_refill = now
1116end
1117
1118local next = current - tokens
1119local stored = current
1120if next >= 0 or overflow then
1121 stored = next
1122end
1123
1124elapsed = math.max(0, now - last_refill)
1125local ttl = math.ceil(math.ceil(limit - stored) * refill_ms - elapsed)
1126if ttl < 1 then ttl = 1 end
1127redis.call("SET", key, stored, "PX", ttl)
1128redis.call("SET", last_refill_key, last_refill, "PX", ttl)
1129-- Use 17 significant digits to round-trip both numbers.
1130return { string.format("%.17g", next), string.format("%.17g", elapsed) }
1131`
1132 }
1133).withReturnType<readonly [remaining: string, elapsedMillis: string]>()
1134
1135const adaptiveConsumeScript = Redis.script(
1136 (key: string, tokens: number, fallbackWindowMillis: number, ttlGraceMillis: number) => [
1137 key,
1138 tokens,
1139 fallbackWindowMillis,
1140 ttlGraceMillis
1141 ],
1142 {
1143 numberOfKeys: 1,
1144 lua: `
1145local key = KEYS[1]
1146local tokens = tonumber(ARGV[1])
1147local fallback_window_ms = tonumber(ARGV[2])
1148local ttl_grace_ms = tonumber(ARGV[3])
1149
1150local time = redis.call("TIME")
1151local now = (tonumber(time[1]) * 1000) + math.floor(tonumber(time[2]) / 1000)
1152
1153local phase = redis.call("HGET", key, "phase")
1154if not phase then
1155 return { 0, 0, "inactive" }
1156end
1157
1158local epoch = tonumber(redis.call("HGET", key, "epoch") or "0")
1159
1160if phase == "cooldown" then
1161 local cooldown_until = tonumber(redis.call("HGET", key, "cooldownUntil") or "0")
1162 if cooldown_until > now then
1163 return { math.ceil(cooldown_until - now), epoch, "cooldown" }
1164 end
1165
1166 epoch = epoch + 1
1167 redis.call(
1168 "HSET",
1169 key,
1170 "phase",
1171 "learning",
1172 "epoch",
1173 epoch,
1174 "cooldownUntil",
1175 0,
1176 "learningStartedAt",
1177 now,
1178 "observedTokens",
1179 tokens,
1180 "learnedLimit",
1181 0,
1182 "learnedWindowMillis",
1183 0
1184 )
1185 redis.call("PEXPIRE", key, math.max(1, math.ceil(fallback_window_ms + ttl_grace_ms)))
1186 return { 0, epoch, "learning" }
1187end
1188
1189if phase == "learning" then
1190 local observed_tokens = tonumber(redis.call("HGET", key, "observedTokens") or "0") + tokens
1191 redis.call("HSET", key, "observedTokens", observed_tokens)
1192 return { 0, epoch, "learning" }
1193end
1194
1195if phase == "learned" then
1196 local cooldown_until = tonumber(redis.call("HGET", key, "cooldownUntil") or "0")
1197 local observed_tokens = tonumber(redis.call("HGET", key, "observedTokens") or "0")
1198 local learned_limit = tonumber(redis.call("HGET", key, "learnedLimit") or "0")
1199 local learned_window_ms = tonumber(redis.call("HGET", key, "learnedWindowMillis") or "0")
1200 local refill_rate_ms = learned_window_ms / learned_limit
1201
1202 if cooldown_until <= now then
1203 observed_tokens = 0
1204 cooldown_until = now
1205 end
1206
1207 observed_tokens = observed_tokens + tokens
1208 cooldown_until = cooldown_until + (refill_rate_ms * tokens)
1209
1210 local ttl = cooldown_until - now
1211 local ttl_total = observed_tokens * refill_rate_ms
1212 local elapsed = ttl_total - ttl
1213 local window_number = math.floor((observed_tokens - 1) / learned_limit)
1214 local remaining = (window_number * learned_window_ms) - elapsed
1215
1216 redis.call("HSET", key, "observedTokens", observed_tokens, "cooldownUntil", cooldown_until)
1217
1218 if remaining <= 0 then
1219 return { 0, epoch, "learned" }
1220 end
1221 return { math.ceil(remaining), epoch, "learned" }
1222end
1223
1224return { 0, epoch, phase }
1225`
1226 }
1227).withReturnType<readonly [delayMillis: number, epoch: number, phase: AdaptivePhase]>()
1228
1229const adaptiveFeedbackScript = Redis.script(
1230 (
1231 key: string,
1232 epoch: number,
1233 tokens: number,
1234 retryAfterMillis: number,
1235 ttlGraceMillis: number,
1236 maxWindowMillis: number
1237 ) => [
1238 key,
1239 epoch,
1240 tokens,
1241 retryAfterMillis,
1242 ttlGraceMillis,
1243 maxWindowMillis
1244 ],
1245 {
1246 numberOfKeys: 1,
1247 lua: `
1248local key = KEYS[1]
1249local feedback_epoch = tonumber(ARGV[1])
1250local tokens = tonumber(ARGV[2])
1251local retry_after_ms = tonumber(ARGV[3])
1252local ttl_grace_ms = tonumber(ARGV[4])
1253local max_window_ms = tonumber(ARGV[5])
1254
1255if retry_after_ms <= 0 then
1256 retry_after_ms = 1
1257end
1258if retry_after_ms > max_window_ms then
1259 retry_after_ms = max_window_ms
1260end
1261
1262local time = redis.call("TIME")
1263local now = (tonumber(time[1]) * 1000) + math.floor(tonumber(time[2]) / 1000)
1264local cooldown_until = now + retry_after_ms
1265
1266local phase = redis.call("HGET", key, "phase")
1267if not phase then
1268 if feedback_epoch ~= 0 then
1269 return 0
1270 end
1271 redis.call(
1272 "HSET",
1273 key,
1274 "phase",
1275 "cooldown",
1276 "epoch",
1277 0,
1278 "cooldownUntil",
1279 cooldown_until,
1280 "learningStartedAt",
1281 0,
1282 "observedTokens",
1283 0,
1284 "learnedLimit",
1285 0,
1286 "learnedWindowMillis",
1287 0
1288 )
1289 redis.call("PEXPIRE", key, math.max(1, math.ceil(retry_after_ms + ttl_grace_ms)))
1290 return 1
1291end
1292
1293local epoch = tonumber(redis.call("HGET", key, "epoch") or "0")
1294if epoch ~= feedback_epoch then
1295 return 0
1296end
1297
1298if phase == "cooldown" then
1299 local current_cooldown_until = tonumber(redis.call("HGET", key, "cooldownUntil") or "0")
1300 if current_cooldown_until > cooldown_until then
1301 cooldown_until = current_cooldown_until
1302 end
1303 redis.call("HSET", key, "cooldownUntil", cooldown_until)
1304 redis.call("PEXPIRE", key, math.max(1, math.ceil((cooldown_until - now) + ttl_grace_ms)))
1305 return 1
1306end
1307
1308if phase == "learning" then
1309 local observed_tokens = tonumber(redis.call("HGET", key, "observedTokens") or "0")
1310 local accepted_tokens = observed_tokens - tokens
1311 if accepted_tokens <= 0 then
1312 redis.call(
1313 "HSET",
1314 key,
1315 "phase",
1316 "cooldown",
1317 "cooldownUntil",
1318 cooldown_until,
1319 "learningStartedAt",
1320 0,
1321 "observedTokens",
1322 0,
1323 "learnedLimit",
1324 0,
1325 "learnedWindowMillis",
1326 0
1327 )
1328 redis.call("PEXPIRE", key, math.max(1, math.ceil(retry_after_ms + ttl_grace_ms)))
1329 return 1
1330 end
1331
1332 local learning_started_at = tonumber(redis.call("HGET", key, "learningStartedAt") or now)
1333 local learned_window_ms = (now - learning_started_at) + retry_after_ms
1334 if learned_window_ms <= 0 then
1335 learned_window_ms = 1
1336 end
1337 if learned_window_ms > max_window_ms then
1338 learned_window_ms = max_window_ms
1339 end
1340 local next_epoch = epoch + 1
1341 redis.call(
1342 "HSET",
1343 key,
1344 "phase",
1345 "learned",
1346 "epoch",
1347 next_epoch,
1348 "cooldownUntil",
1349 learning_started_at + learned_window_ms,
1350 "observedTokens",
1351 accepted_tokens,
1352 "learnedLimit",
1353 accepted_tokens,
1354 "learnedWindowMillis",
1355 learned_window_ms
1356 )
1357 redis.call("PEXPIRE", key, math.max(1, math.ceil(learned_window_ms + ttl_grace_ms)))
1358 return 1
1359end
1360
1361if phase == "learned" then
1362 redis.call(
1363 "HSET",
1364 key,
1365 "phase",
1366 "cooldown",
1367 "cooldownUntil",
1368 cooldown_until,
1369 "learningStartedAt",
1370 0,
1371 "observedTokens",
1372 0,
1373 "learnedLimit",
1374 0,
1375 "learnedWindowMillis",
1376 0
1377 )
1378 redis.call("PEXPIRE", key, math.max(1, math.ceil(retry_after_ms + ttl_grace_ms)))
1379 return 1
1380end
1381
1382return 0
1383`
1384 }
1385).withReturnType<number>()
1386
1387/**
1388 * Provides a Redis-backed `RateLimiterStore` using `makeStoreRedis`.
1389 *
1390 * @stability unstable
1391 * @category layers
1392 * @since 4.0.0
1393 */
1394export const layerStoreRedis: (
1395 options?: { readonly prefix?: string | undefined }
1396) => Layer.Layer<
1397 RateLimiterStore,
1398 never,
1399 Redis.Redis
1400> = flow(makeStoreRedis, Layer.effect(RateLimiterStore))
1401
1402/**
1403 * Provides a Redis-backed `RateLimiterStore` from wrapped configuration
1404 * options.
1405 *
1406 * @stability unstable
1407 * @category layers
1408 * @since 4.0.0
1409 */
1410export const layerStoreRedisConfig = (
1411 options: Config.Wrap<{ readonly prefix?: string | undefined }>
1412): Layer.Layer<RateLimiterStore, Config.ConfigError, Redis.Redis> =>
1413 Layer.effect(
1414 RateLimiterStore,
1415 Effect.flatMap(Config.unwrap(options), makeStoreRedis)
1416 )