Skip to content

Second-typed RetryAfterSeconds and QuotaTerm (review 78356) #13655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gunbai-bot merged 1 commit into from
Oct 9, 2026
Merged

Second-typed RetryAfterSeconds and QuotaTerm (review 78356) #13655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions dag/extdeps/openrouter/openrouter.dag
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ module extdeps.openrouter.openrouter
import std.types { String, Bool, Int, List, NonEmptyStr }
import std.algebra { trim }
import std.nat { Nat, nat_range_inclusive }
import std.measure { Second, second, second_count }
import std.checked_arithmetic { checked_int_to_nat }
import std.decl_ref { DeclarationRef, WholeDeclaration }
import extdeps.external_authority { ExternalAuthority, ExternalModelScope, ExternalSubjectRef }
Expand Down Expand Up @@ -411,7 +412,7 @@ fn openrouter_limit_source_meaning(s: OpenRouterLimitSource) -> String {
// computes its wait from Retry-After alone; the reset is evidence for the operator, with a live reading
// the thing that would ground its unit.
type OpenRouterRetryAfter
= RetryAfterSeconds { seconds: Nat }
= RetryAfterSeconds { seconds: Second }
| RetryAfterUnparsed { text: String }
| RetryAfterAbsent
| RetryAfterUnread { cause: String }
Expand Down Expand Up @@ -499,7 +500,7 @@ fn openrouter_retry_after_from_digits(text: String) -> OpenRouterRetryAfter {
Present { value: n } =>
match checked_int_to_nat(n: n) {
Absent => RetryAfterUnparsed { text: text }
Present { value: sec } => RetryAfterSeconds { seconds: sec }
Present { value: sec } => RetryAfterSeconds { seconds: second(count: sec) }
}
}
}
Expand Down Expand Up @@ -545,7 +546,7 @@ fn openrouter_decode_rate_limited(body: String, header_dump: OpenRouterHeaderDum

fn openrouter_retry_after_wire(r: OpenRouterRetryAfter) -> String {
match r {
RetryAfterSeconds { seconds: n } => join(["Retry-After ", to_string(n), " s"], "")
RetryAfterSeconds { seconds: n } => join(["Retry-After ", to_string(second_count(s: n)), " s"], "")
RetryAfterUnparsed { text: t } => join(["Retry-After '", t, "' (not delta-seconds; not interpreted)"], "")
RetryAfterAbsent => "no Retry-After"
RetryAfterUnread { cause: c } => join(["Retry-After unknown: the response headers could not be read (", c, ")"], "")
Expand Down
24 changes: 16 additions & 8 deletions dag/gunbc/fabric/fabric_quota.dag
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ module gunbc.fabric_quota
import std.types { String, Bool, Int, NonEmptyStr, List, EpochSecs }
import std.nat { Nat }
import std.checked_arithmetic { checked_int_to_nat }
import std.measure { Measure }
import std.measure { Measure, Second, second, second_count }
import extdeps.api_rate_limit { UpstreamRateLimit, upstream_rate_limit_key }
import product.capacity.quota { quota_partition, quota_root_pool }
import product.capacity.event_chain { PartitionId }
Expand Down Expand Up @@ -41,19 +41,26 @@ fn quota_policy(term_seconds: Nat) -> LeasePolicy {
// at second 59 of a minute window only one second is left. The window end is read from the SAME
// clock reading the lease is stamped with, so the term fits by construction rather than by luck.
type QuotaTerm
= QuotaTermSeconds { seconds: Nat }
= QuotaTermSeconds { seconds: Second }
| QuotaTermToWindowEnd

fn quota_term_to_window_end(limit: UpstreamRateLimit, at: EpochSecs) -> Nat? {
fn quota_term_to_window_end(limit: UpstreamRateLimit, at: EpochSecs) -> Second? {
match pool_window_at(pool: quota_root_pool(limit: limit), at: at) {
WindowResolved { window: w } => if w.end > at { checked_int_to_nat(n: w.end - at) } else { none }
WindowResolved { window: w } => if w.end > at { checked_second(n: w.end - at) } else { none }
WholeLifetime => none
InstantBeforeAnchor { anchor: _, at: _ } => none
}
}

// THE INSTANT THE CURRENT WINDOW BEGAN, read off the same pool the term is: what a caller needs to tell
// which of its outstanding sends were admitted in an earlier window.
fn checked_second(n: Int) -> Second? {
match checked_int_to_nat(n: n) {
Absent => none
Present { value: v } => Present { value: second(count: v) }
}
}

fn quota_window_start(limit: UpstreamRateLimit, at: EpochSecs) -> EpochSecs? {
match pool_window_at(pool: quota_root_pool(limit: limit), at: at) {
WindowResolved { window: w } => Present { value: w.start }
Expand All @@ -62,15 +69,15 @@ fn quota_window_start(limit: UpstreamRateLimit, at: EpochSecs) -> EpochSecs? {
}
}

fn quota_term_seconds(limit: UpstreamRateLimit, term: QuotaTerm, at: EpochSecs) -> Nat? {
fn quota_term_seconds(limit: UpstreamRateLimit, term: QuotaTerm, at: EpochSecs) -> Second? {
match term {
QuotaTermSeconds { seconds: n } => Present { value: n }
QuotaTermToWindowEnd => quota_term_to_window_end(limit: limit, at: at)
}
}

fn fabric_quota_lease(short_hostname: String, limit: UpstreamRateLimit, amount: Nat, actor: NonEmptyStr, term_seconds: Nat) -> QuotaLease {
fabric_quota_lease_for(short_hostname: short_hostname, limit: limit, amount: amount, actor: actor, term: QuotaTermSeconds { seconds: term_seconds })
fabric_quota_lease_for(short_hostname: short_hostname, limit: limit, amount: amount, actor: actor, term: QuotaTermSeconds { seconds: second(count: term_seconds) })
}

fn fabric_quota_lease_for(short_hostname: String, limit: UpstreamRateLimit, amount: Nat, actor: NonEmptyStr, term: QuotaTerm) -> QuotaLease {
Expand All @@ -91,7 +98,8 @@ fn fabric_quota_lease_for(short_hostname: String, limit: UpstreamRateLimit, amou
fn fabric_quota_lease_at(store: FabricStorageBinding, now: EpochSecs, limit: UpstreamRateLimit, amount: Nat, actor: NonEmptyStr, term: QuotaTerm) -> QuotaLease {
match quota_term_seconds(limit: limit, term: term, at: now) {
Absent => QuotaLeaseRefused { detail: join([upstream_rate_limit_key(l: limit) as String, ": no window end is defined at this instant"], "") }
Present { value: term_seconds } => {
Present { value: term } => {
let term_seconds = second_count(s: term)
let partition = quota_partition(limit: limit)
match fabric_seat_acquire(
store: store, partition: partition, root: quota_root_pool(limit: limit), actor: actor,
Expand Down Expand Up @@ -130,7 +138,7 @@ fn fabric_quota_observe_upstream(short_hostname: String, limit: UpstreamRateLimi
let partition = quota_partition(limit: limit)
match fabric_pool_transition(
store: store, partition: partition, root: quota_root_pool(limit: limit), actor: actor, at: now,
payload: fn(identity) { PoolUpstreamObserved { remaining: remaining, term_seconds: term } }, attempts: 3, budget: 4096,
payload: fn(identity) { PoolUpstreamObserved { remaining: remaining, term_seconds: second_count(s: term) } }, attempts: 3, budget: 4096,
) {
PoolTransitionAppended { id: _ } => QuotaUpstreamRecorded { partition: partition, remaining: remaining }
other => QuotaObservationRefused { detail: pool_transition_wire(t: other) }
Expand Down
20 changes: 10 additions & 10 deletions dag/test/claim/openrouter_hosted_route_witness_test.dag
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,7 @@ fn w_charge_admitted(limit: UpstreamRateLimit, at: Int) -> Bool {
match quota_term_to_window_end(limit: limit, at: at) {
Absent => false
Present { value: term } =>
match pool_acquire(pool: quota_root_pool(limit: limit), reference: "charge" as NonEmptyStr, amount: Measure { count: 1 }, at: at, term_seconds: term) {
match pool_acquire(pool: quota_root_pool(limit: limit), reference: "charge" as NonEmptyStr, amount: Measure { count: 1 }, at: at, term_seconds: second_count(s: term)) {
PoolAdvanced { pool: _ } => true
PoolRefused { refusal: _ } => false
}
Expand Down Expand Up @@ -315,7 +315,7 @@ fn w_observed(pool: Pool<Count, One>, remaining: Nat, at: Int) -> Pool<Count, On
match quota_term_to_window_end(limit: day, at: at) {
Absent => none
Present { value: term } =>
match observe_upstream_remaining(pool: pool, remaining: Measure { count: remaining }, at: at, term_seconds: term, reading: join(["witness@", to_string(at)], "") as NonEmptyStr) {
match observe_upstream_remaining(pool: pool, remaining: Measure { count: remaining }, at: at, term_seconds: second_count(s: term), reading: join(["witness@", to_string(at)], "") as NonEmptyStr) {
PoolAdvanced { pool: p } => Present { value: p }
PoolRefused { refusal: _ } => none
}
Expand Down Expand Up @@ -383,7 +383,7 @@ test fn a_settlement_after_rollover_is_refused_before_it_could_poison_the_partit
match quota_term_to_window_end(limit: minute, at: at) {
Absent => false
Present { value: term } => {
let acquired = w_env(id: "e1", at: at, payload: PoolAcquired { reference: "charge" as NonEmptyStr, amount: 1, term_seconds: term })
let acquired = w_env(id: "e1", at: at, payload: PoolAcquired { reference: "charge" as NonEmptyStr, amount: 1, term_seconds: second_count(s: term) })
match pool_fold(root: quota_root_pool(limit: minute), oldest_first: [acquired]) {
PoolFoldRefused { at_event: _, wire: _ } => false
PoolFolded { pool: p, generation: _ } =>
Expand Down Expand Up @@ -414,8 +414,8 @@ test fn two_decreasing_upstream_readings_in_one_second_fold_to_the_lower() -> Bo
Absent => false
Present { value: term } =>
match pool_fold(root: quota_root_pool(limit: day), oldest_first: [
w_env(id: "o1", at: at, payload: PoolUpstreamObserved { remaining: 3, term_seconds: term }),
w_env(id: "o2", at: at, payload: PoolUpstreamObserved { remaining: 1, term_seconds: term }),
w_env(id: "o1", at: at, payload: PoolUpstreamObserved { remaining: 3, term_seconds: second_count(s: term) }),
w_env(id: "o2", at: at, payload: PoolUpstreamObserved { remaining: 1, term_seconds: second_count(s: term) }),
]) {
PoolFoldRefused { at_event: _, wire: _ } => false
PoolFolded { pool: p, generation: g } => g == 2 && w_day_headroom(pool: p, at: at) == 1
Expand Down Expand Up @@ -463,7 +463,7 @@ fn w_round_admitted(minute: Pool<Count, One>, carried: Nat, at: Int) -> Bool {
match quota_term_to_window_end(limit: openrouter_free_models_per_minute, at: at) {
Absent => false
Present { value: term } =>
match pool_acquire(pool: minute, reference: "this-round" as NonEmptyStr, amount: Measure { count: carried + 1 }, at: at, term_seconds: term) {
match pool_acquire(pool: minute, reference: "this-round" as NonEmptyStr, amount: Measure { count: carried + 1 }, at: at, term_seconds: second_count(s: term)) {
PoolAdvanced { pool: _ } => true
PoolRefused { refusal: _ } => false
}
Expand Down Expand Up @@ -534,7 +534,7 @@ test fn a_429_decodes_to_the_typed_refusal_with_the_upstreams_retry_hints() -> B
match openrouter_decode_rate_limited(body: w_429_body, header_dump: HeaderDumpRead { text: w_429_headers }) {
Absent => false
Present { value: r } =>
(match r.hints.retry_after { RetryAfterSeconds { seconds: n } => n == 30 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
(match r.hints.retry_after { RetryAfterSeconds { seconds: n } => second_count(s: n) == 30 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
&& (match r.hints.reset { Present { value: v } => v == "1791265800000" Absent => false })
&& (match r.hints.remaining { Present { value: v } => v == "0" Absent => false })
&& (match r.provider_code { Absent => true Present { value: _ } => false })
Expand All @@ -552,7 +552,7 @@ test fn a_retry_after_that_is_not_seconds_and_a_missing_hint_are_stated_not_gues
Absent => false
})
&& (match openrouter_decode_rate_limited(body: w_429_body, header_dump: HeaderDumpRead { text: "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 12\r\nX-RateLimit-Reset: 5\r\n\r\n" }) {
Present { value: r } => (match r.hints.retry_after { RetryAfterSeconds { seconds: n } => n == 12 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
Present { value: r } => (match r.hints.retry_after { RetryAfterSeconds { seconds: n } => second_count(s: n) == 12 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
&& (match r.hints.reset { Present { value: v } => v == "5" Absent => false })
Absent => false
})
Expand All @@ -561,7 +561,7 @@ test fn a_retry_after_that_is_not_seconds_and_a_missing_hint_are_stated_not_gues
Absent => false
})
&& (match openrouter_decode_rate_limited(body: w_429_body, header_dump: HeaderDumpRead { text: "HTTP/1.1 100 Continue\r\nretry-after: 99\r\n\r\nHTTP/2 429\r\nretry-after: 7\r\n\r\n" }) {
Present { value: r } => (match r.hints.retry_after { RetryAfterSeconds { seconds: n } => n == 7 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
Present { value: r } => (match r.hints.retry_after { RetryAfterSeconds { seconds: n } => second_count(s: n) == 7 RetryAfterUnparsed { text: _ } => false RetryAfterAbsent => false RetryAfterUnread { cause: _ } => false })
Absent => false
})
}
Expand Down Expand Up @@ -605,7 +605,7 @@ test fn a_signed_fractional_or_suffixed_retry_after_is_not_read_as_seconds() ->
&& w_is_unparsed(r: w_retry_after_of(value: "-30"))
&& w_is_unparsed(r: w_retry_after_of(value: "3.5"))
&& w_is_unparsed(r: w_retry_after_of(value: "30s"))
&& (match w_retry_after_of(value: "30") { Present { value: RetryAfterSeconds { seconds: n } } => n == 30 Present { value: _ } => false Absent => false })
&& (match w_retry_after_of(value: "30") { Present { value: RetryAfterSeconds { seconds: n } } => second_count(s: n) == 30 Present { value: _ } => false Absent => false })
}

// RED: a 402, or any body that is not a 429, is NOT a rate-limit refusal -- it keeps its own typed error
Expand Down