nx_rate_limiter.nx source
↩ module page · 199 lines · 8150 B
1// nx_rate_limiter.nx -- token-bucket rate limiter for NX-INGEST.
2//
3// module: nishi-core.ingest.rate_limiter
4// depends: nishi-core.io.syscalls, nishi-core.io.iso8601
5// disk_kb: 4
6// capability: CORE_IO
7//
8// license_tier: PUBLIC_NISHI_SUBSTRATE
9// genealogy_id: token_bucket_algorithm_floyd_jacobson_1992 +
10// rfc_6585_http_429_too_many_requests +
11// aws_token_bucket_rate_limiter_pattern +
12// nishi_ingestion_s_class_cardinal_2026
13//
14// Polite-pool rate limiter for every ingestion source. Token bucket
15// per RFC 6585 + Floyd-Jacobson 1992. Substrate Cardinal 14 graceful
16// degradation: when bucket is empty, fetch waits or yields THROTTLED
17// verdict (caller decides between block-and-wait vs return-control).
18//
19// ===== Why token bucket =========================================
20//
21// Token bucket gives both:
22// - average rate enforcement (replenish at rate R)
23// - burst tolerance (capacity B allows short bursts up to B
24// requests when upstream is healthy)
25//
26// Leaky bucket would enforce strict rate but disallow bursts —
27// worse for real-world polite-pool patterns where upstreams
28// tolerate burst-then-pause.
29//
30// ===== Polite-pool defaults =====================================
31//
32// Per the nishi-library sources.toml convention + RFC 6585:
33// - arXiv: 1 req / 3 sec (cap 3 tokens; refill 1 every 3s)
34// - bioRxiv / medRxiv: 1 req / sec (cap 5; refill 1 every 1s)
35// - USDA APIs: 1 req / sec (cap 5; refill 1 every 1s)
36// - Kew SID: 0.5 req / sec (cap 2; refill 1 every 2s)
37// - GBIF: 1 req / sec
38// - OpenFarm: 1 req / sec
39//
40// All values in the source-adapter's SourceDescriptor.
41
42// nx_safety_envelope:
43// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
44// sil_target: SIL1
45// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
46// verdict: NOT_YET_EVALUATED
47
48import "nx_syscalls.nx"
49import "nx_iso8601.nx"
50
51// ===== Verdict ====================================================
52
53const NX_RATE_ALLOWED: i64 = 1
54const NX_RATE_THROTTLED: i64 = 2 // current bucket empty; wait + retry
55const NX_RATE_EXHAUSTED: i64 = 3 // burst window AND refill rate exhausted
56const NX_RATE_INVALID: i64 = 4 // misconfigured limiter
57
58func nx_rate_verdict_name(v: i64) -> *u8 {
59 if v == NX_RATE_ALLOWED { return "ALLOWED" }
60 if v == NX_RATE_THROTTLED { return "THROTTLED" }
61 if v == NX_RATE_EXHAUSTED { return "EXHAUSTED" }
62 if v == NX_RATE_INVALID { return "INVALID" }
63 return "UNKNOWN"
64}
65
66// ===== RateLimiter struct =========================================
67
68struct RateLimiter {
69 limiter_hk: i64,
70 descriptor_hk: i64, // FK to SourceDescriptor
71 capacity: i64, // max tokens (burst tolerance)
72 refill_rate_per_sec_q10: i64, // tokens added per second (Q10)
73 current_tokens_q10: i64, // current bucket level (Q10)
74 last_refill_unix: i64, // when bucket was last topped up
75 last_request_unix: i64, // diagnostic; last attempt time
76 n_allowed: i64,
77 n_throttled: i64,
78 n_exhausted: i64,
79}
80
81const NX_RATE_LIMITER_BYTES: i64 = 80 // 10 fields * 8 bytes
82
83// ===== Constructor ================================================
84
85func nx_rate_limiter_new(
86 descriptor_hk: i64,
87 capacity: i64,
88 refill_rate_per_sec_q10: i64,
89 now_unix: i64
90) -> *RateLimiter {
91 if capacity <= 0 { return 0 as *RateLimiter }
92 if refill_rate_per_sec_q10 <= 0 { return 0 as *RateLimiter }
93 let raw: *u8 = sys_mmap(NX_RATE_LIMITER_BYTES)
94 let r: *RateLimiter = raw as *RateLimiter
95 r.limiter_hk = 0
96 r.descriptor_hk = descriptor_hk
97 r.capacity = capacity
98 r.refill_rate_per_sec_q10 = refill_rate_per_sec_q10
99 r.current_tokens_q10 = capacity * 1024 // start full
100 r.last_refill_unix = now_unix
101 r.last_request_unix = now_unix
102 r.n_allowed = 0
103 r.n_throttled = 0
104 r.n_exhausted = 0
105 return r
106}
107
108// ===== Refill (top up bucket based on elapsed time) ==============
109//
110// Called before every request attempt. Refills the bucket based on
111// (now - last_refill_unix) * refill_rate, capped at capacity.
112
113func nx_rate_limiter_refill(r: *RateLimiter, now_unix: i64) -> i64 {
114 if r == 0 as *RateLimiter { return NX_RATE_INVALID }
115 let elapsed_sec: i64 = now_unix - r.last_refill_unix
116 if elapsed_sec <= 0 { return NX_RATE_ALLOWED }
117 let tokens_added_q10: i64 = elapsed_sec * r.refill_rate_per_sec_q10
118 let max_tokens_q10: i64 = r.capacity * 1024
119 var new_level_q10: i64 = r.current_tokens_q10 + tokens_added_q10
120 if new_level_q10 > max_tokens_q10 { new_level_q10 = max_tokens_q10 }
121 r.current_tokens_q10 = new_level_q10
122 r.last_refill_unix = now_unix
123 return NX_RATE_ALLOWED
124}
125
126// ===== Try-acquire (consume one token if available) ==============
127
128const NX_RATE_TOKEN_COST_Q10: i64 = 1024 // one full token per request
129
130func nx_rate_limiter_try_acquire(r: *RateLimiter, now_unix: i64) -> i64 {
131 if r == 0 as *RateLimiter { return NX_RATE_INVALID }
132 nx_rate_limiter_refill(r, now_unix)
133 r.last_request_unix = now_unix
134 if r.current_tokens_q10 >= NX_RATE_TOKEN_COST_Q10 {
135 r.current_tokens_q10 = r.current_tokens_q10 - NX_RATE_TOKEN_COST_Q10
136 r.n_allowed = r.n_allowed + 1
137 return NX_RATE_ALLOWED
138 }
139 // No token available; check if we're past total-exhaustion
140 // threshold (consecutive empty buckets over many minutes).
141 if r.n_throttled > 1000 {
142 r.n_exhausted = r.n_exhausted + 1
143 return NX_RATE_EXHAUSTED
144 }
145 r.n_throttled = r.n_throttled + 1
146 return NX_RATE_THROTTLED
147}
148
149// ===== Wait-until-allowed (compute next acquire-time) =============
150//
151// Returns the unix timestamp at which the bucket will have at least
152// one token available. Caller can sleep until this time then retry.
153
154func nx_rate_limiter_next_allowed_unix(r: *RateLimiter, now_unix: i64) -> i64 {
155 if r == 0 as *RateLimiter { return now_unix }
156 nx_rate_limiter_refill(r, now_unix)
157 if r.current_tokens_q10 >= NX_RATE_TOKEN_COST_Q10 { return now_unix }
158 let tokens_needed_q10: i64 = NX_RATE_TOKEN_COST_Q10 - r.current_tokens_q10
159 if r.refill_rate_per_sec_q10 <= 0 { return now_unix + 86400 } // never if rate is 0
160 let wait_seconds: i64 = (tokens_needed_q10 + r.refill_rate_per_sec_q10 - 1) / r.refill_rate_per_sec_q10
161 return now_unix + wait_seconds
162}
163
164// ===== Burst-mode acquire (consume multiple tokens at once) =======
165//
166// Some sources support burst patterns (e.g. parallel-fetch). Caller
167// requests N tokens; substrate either grants all-or-none.
168
169func nx_rate_limiter_try_acquire_n(r: *RateLimiter, n: i64, now_unix: i64) -> i64 {
170 if r == 0 as *RateLimiter { return NX_RATE_INVALID }
171 if n <= 0 { return NX_RATE_INVALID }
172 if n > r.capacity { return NX_RATE_INVALID } // can't exceed burst capacity
173 nx_rate_limiter_refill(r, now_unix)
174 r.last_request_unix = now_unix
175 let needed_q10: i64 = n * NX_RATE_TOKEN_COST_Q10
176 if r.current_tokens_q10 >= needed_q10 {
177 r.current_tokens_q10 = r.current_tokens_q10 - needed_q10
178 r.n_allowed = r.n_allowed + n
179 return NX_RATE_ALLOWED
180 }
181 r.n_throttled = r.n_throttled + 1
182 return NX_RATE_THROTTLED
183}
184
185// ===== HTTP 429 response handler ===================================
186//
187// Per RFC 6585: when upstream returns 429 Too Many Requests,
188// substrate respects any Retry-After header by stalling the
189// limiter for that many seconds. Composes against the HTTP
190// response parser.
191
192func nx_rate_limiter_apply_retry_after(r: *RateLimiter, retry_after_seconds: i64, now_unix: i64) -> i64 {
193 if r == 0 as *RateLimiter { return NX_RATE_INVALID }
194 if retry_after_seconds <= 0 { return NX_RATE_INVALID }
195 // Drain the bucket + delay last_refill by retry_after
196 r.current_tokens_q10 = 0
197 r.last_refill_unix = now_unix + retry_after_seconds
198 return NX_RATE_ALLOWED
199}