pollrow.bend

cswap-bend-cli@0.1.0.0 · 229 lines · 36 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/json.bend as Json
3import cswap-bend-cli@0.1.0.0/clock.bend as Clock
4import cswap-bend-cli@0.1.0.0/policy/usage.bend as PUsage
5import cswap-bend-cli@0.1.0.0/policy/types.bend as PT
6import cswap-bend-cli@0.1.0.0/poll.bend as Poll
7
8# A stored usage.json row as the planner reads it: failures, claims, holds, trust and
9# eligibility to fetch (the rules of src/policy where it has them).
10
11# The fields of one usage.json row that planning and display read.
12type Row is Data:
13  Row{fetched: Maybe<&2, U32>, last_good: Json.Json, last_attempt: Maybe<&2, U32>, failures: U32,
14      last_error: Maybe<&2, String>, backoff_until: Maybe<&2, U32>, next_poll: Maybe<&2, U32>,
15      interval: Maybe<&2, U32>, last429: Maybe<&2, U32>, strikes: U32, struck_fp: Maybe<&2, String>,
16      claim_until: Maybe<&2, U32>, held_until: Maybe<&2, U32>}
17
18def is_err(m: Maybe<&2, String>, +s: String) -> Bool:
19  match m:
20    case Some{x}:
21      String.eq(x, s)
22    case None{}:
23      False{}
24
25# A 429 on this token is recent enough to keep the post-429 cadence: measured
26# from when the honoured backoff lifts, for one hour.
27def recent_429.anchor(+l: U32, bu: Maybe<&2, U32>, is429: Bool) -> U32:
28  match bu is429:
29    case Some{b} True{}:
30      Poll.max_u(l, b)
31    case _ _:
32      l
33
34def recent.of(l: Maybe<&2, U32>, bu: Maybe<&2, U32>, is429: Bool, +now: U32) -> Bool:
35  match l:
36    case None{}:
37      False{}
38    case Some{+x}:
39      U32.is_lt(now, (recent_429.anchor(x, bu, is429) + Poll.recent_429_window() : U32))
40
41def recent_429(+r: Row, +now: U32) -> Bool:
42  match r:
43    case Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
44      recent.of(l429, bu, is_err(le, "http-429"), now)
45
46def num.int(ps: List<&2, String>) -> Maybe<&2, U32>:
47  match ps:
48    case Con{i, t}:
49      U32.read(i)
50    case Nil{}:
51      None{}
52
53# An integer second count from a JSON number (`1792740000.1` -> 1792740000).
54def num.u(j: Json.Json) -> Maybe<&2, U32>:
55  match j:
56    case Json.JNum{t}:
57      num.int(String.split(t, '.'))
58    case _:
59      None{}
60
61def jget(+k: String, +row: Json.Json) -> Json.Json:
62  Maybe.default(&2, Json.Json, Json.get(k, row), Json.JNull{})
63
64def num.field(+k: String, +row: Json.Json) -> Maybe<&2, U32>:
65  num.u(jget(k, row))
66
67def str.field(+k: String, +row: Json.Json) -> Maybe<&2, String>:
68  Json.as_string(jget(k, row))
69
70def obj_or_null(j: Json.Json) -> Json.Json:
71  match j:
72    case Json.JObj{f}:
73      Json.JObj{f}
74    case _:
75      Json.JNull{}
76
77def last_good.of(+row: Json.Json) -> Json.Json:
78  obj_or_null(jget("lastGood", row))
79
80def row.of(+row: Json.Json) -> Row:
81  Row{num.field("fetchedAt", row), last_good.of(row), num.field("lastAttemptAt", row),
82    Maybe.default(&2, U32, num.field("consecutiveFailures", row), 0), str.field("lastError", row),
83    num.field("backoffUntil", row), num.field("nextPollAt", row), num.field("pollIntervalS", row),
84    num.field("last429At", row), Maybe.default(&2, U32, num.field("authDeadStrikes", row), 0),
85    str.field("struckFingerprint", row), num.field("claimUntil", row), num.field("heldUntil", row)}
86
87def before.m(m: Maybe<&2, U32>, +now: U32) -> Bool:
88  match m:
89    case Some{x}:
90      U32.is_lt(now, x)
91    case None{}:
92      False{}
93
94def legacy_claim(m: Maybe<&2, U32>, +now: U32) -> Bool:
95  match m:
96    case Some{a}:
97      U32.is_lt(Poll.sub0(now, a), 10)
98    case None{}:
99      False{}
100
101# A fetch lease is live: claimUntil when present, else the legacy 10 s after
102# lastAttemptAt.
103def live_claim(claim_until: Maybe<&2, U32>, last_attempt: Maybe<&2, U32>, +now: U32) -> Bool:
104  match claim_until:
105    case Some{c}:
106      U32.is_lt(now, c)
107    case None{}:
108      legacy_claim(last_attempt, now)
109
110def age_le(age: Maybe<&2, U32>, +max: U32) -> Bool:
111  match age:
112    case None{}:
113      False{}
114    case Some{a}:
115      U32.is_le(a, max)
116
117def age_of(fetched: Maybe<&2, U32>, +now: U32) -> Maybe<&2, U32>:
118  match fetched:
119    case Some{f}:
120      Some{Poll.sub0(now, f)}
121    case None{}:
122      None{}
123
124def err_kind(rl: Bool) -> PT.ErrKind:
125  match rl:
126    case True{}:
127      PT.ErrHttp429{}
128    case False{}:
129      PT.ErrOther{}
130
131# Seconds to stay in backoff after the n-th failure: src/policy's backoff_for
132# (30 s * 2^(n-1) capped at 600 s; a Retry-After raises it).
133def failure_backoff(+failures: U32, retry_after: Maybe<&2, U32>, rate_limited: Bool) -> U32:
134  U32.from_nat(PUsage.backoff_for(U32.to_nat(failures), err_kind(rate_limited), Poll.nat_maybe(retry_after)))
135
136def nat_of(m: Maybe<&2, U32>, +d: U32) -> Nat:
137  U32.to_nat(Maybe.default(&2, U32, m, d))
138
139def blocked(r: Row, +now: U32) -> Bool:
140  match r:
141    case Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
142      U32.is_ge(st, 1) || before.m(bu, now) || before.m(hu, now) || live_claim(cu, la, now)
143
144def mode.pick(+respect: Bool, +repair: Bool, +stale: Bool, +due: Bool, +none: Bool, +over: Bool) -> Bool:
145  match respect repair:
146    case True{} _:
147      stale && (due || none || over)
148    case False{} True{}:
149      due || (stale && (none || over))
150    case False{} False{}:
151      due || stale
152
153def mode.go(+respect: Bool, +repair: Bool, +fa: Maybe<&2, Nat>, +np: Maybe<&2, Nat>, +pi: Nat, +now: Nat) -> Bool:
154  mode.pick(respect, repair, PUsage.poll.stale(fa, now, 180n), PUsage.poll.planned_due(np, now), PUsage.poll.unplanned(np), repair && PUsage.poll.overslept(np, now, pi))
155
156def mode.row(r: Row, +now: U32, +respect: Bool, +repair: Bool) -> Bool:
157  match r:
158    case Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
159      mode.go(respect, repair, Poll.nat_maybe(fetched), Poll.nat_maybe(np), nat_of(iv, 600), U32.to_nat(now))
160
161# Whether a stored row may be fetched now: not dead, in backoff, held or
162# claimed, then by caller mode (on-demand `respect` = stale and due; the
163# scheduled tick is policy's poll_due).
164def row_eligible(+r: Row, +now: U32, +respect_plans: Bool, +repair: Bool) -> Bool:
165  Bool.not(blocked(r, now)) && mode.row(r, now, respect_plans, repair)
166
167def nat0(m: Maybe<&2, U32>) -> Nat:
168  match m:
169    case Some{x}:
170      U32.to_nat(x)
171    case None{}:
172      0n
173
174def err_pick(rl: Bool) -> PT.ErrKind:
175  match rl:
176    case True{}:
177      PT.ErrHttp429{}
178    case False{}:
179      PT.ErrOther{}
180
181def kind_of_err(m: Maybe<&2, String>) -> PT.ErrKind:
182  match m:
183    case Some{e}:
184      err_pick(String.eq(e, "http-429"))
185    case None{}:
186      PT.ErrNone{}
187
188def legacy_until(la: Maybe<&2, U32>) -> Nat:
189  match la:
190    case Some{a}:
191      U32.to_nat((a + 10 : U32))
192    case None{}:
193      0n
194
195# A claim lease: claimUntil, else the legacy 10 s after lastAttemptAt.
196def lease_until(cu: Maybe<&2, U32>, la: Maybe<&2, U32>) -> Nat:
197  match cu:
198    case Some{c}:
199      U32.to_nat(c)
200    case None{}:
201      legacy_until(la)
202
203def policy_with(+u: PT.Usage, +fetched: Maybe<&2, U32>, +bu: Maybe<&2, U32>, +le: Maybe<&2, String>, +nf: U32, +np: Maybe<&2, U32>, +iv: Maybe<&2, U32>, +st: U32, +hu: Maybe<&2, U32>, +lease: Nat) -> PT.Usage:
204  match u:
205    case PT.Usage{slot, five, fr, seven, sr, ss, fa, bo, er, nfl, npp, pi, ad, hh, cc, se}:
206      PT.Usage{slot, five, fr, seven, sr, ss, Poll.nat_maybe(fetched), nat0(bu), kind_of_err(le), U32.to_nat(nf), Poll.nat_maybe(np), nat0(iv), U32.to_nat(st), nat0(hu), lease, PT.NoSentinel{}}
207
208# The policy core's view of a stored row: its windows, freshness and holds.
209def policy_row(+r: Row) -> PT.Usage:
210  match r:
211    case Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
212      policy_with(Poll.policy_usage(lg), fetched, bu, le, fl, np, iv, st, hu, lease_until(cu, la))
213
214def decision.pick(ok: Bool, lg: Json.Json) -> Json.Json:
215  match ok:
216    case True{}:
217      lg
218    case False{}:
219      Json.JNull{}
220
221def last_good_of(+r: Row) -> Json.Json:
222  match r:
223    case Row{fetched, lg, la, fl, le, bu, np, iv, l429, st, sfp, cu, hu}:
224      lg
225
226# The value switch decisions run on: last-good while policy trusts it (at most
227# 300 s old, or deliberately stale and inside the cap), JNull when unknown.
228def decision_usage(+r: Row, models: List<&2, String>, +now: U32) -> Json.Json:
229  decision.pick(PUsage.trusted(policy_row(r), U32.to_nat(now), models), last_good_of(r))