urec.bend

cswap-bend-cli@0.1.0.0 · 246 lines · 43 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/pollrow.bend as Pollrow
3import cswap-bend-cli@0.1.0.0/fs/write.bend as FWrite
4import cswap-bend-cli@0.1.0.0/json.bend as Json
5import cswap-bend-cli@0.1.0.0/fs.bend as Fs
6import cswap-bend-cli@0.1.0.0/act.bend as Act
7import cswap-bend-cli@0.1.0.0/lock.bend as Lock
8import cswap-bend-cli@0.1.0.0/poll.bend as Poll
9import cswap-bend-cli@0.1.0.0/ustore.bend as Ustore
10
11# One finished fetch. kind: 0 success, 1 failure, 2 sentinel (nothing to store
12# but the lease). `struck` is the fingerprint of the refresh token the server
13# rejected for good (JNull otherwise).
14type Rec is Data:
15  Rec{id: Ustore.Id, claim: String, kind: U32, usage: Json.Json, error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>}
16
17def release(+row: Json.Json) -> Json.Json:
18  Json.set("claimUntil", Json.JNum{"0.0"}, Json.set("claimId", Json.JNull{}, row))
19
20def with_plan(m: Maybe<&2, Poll.Sched>, +row: Json.Json) -> Json.Json:
21  match m:
22    case None{}:
23      row
24    case Some{Poll.Sched{next, interval}}:
25      Json.set("pollIntervalS", Ustore.fl(interval), Json.set("nextPollAt", Ustore.fl(next), row))
26
27def apply_ok(+row: Json.Json, +usage: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
28  Json.set("authDeadStrikes", Json.JNum{"0"},
29    Json.set("rejectedFingerprint", Json.JNull{},
30      Json.set("backoffUntil", Json.JNull{},
31        Json.set("lastError", Json.JNull{},
32          Json.set("consecutiveFailures", Json.JNum{"0"},
33            with_plan(plan, Json.set("fetchedAt", Ustore.fl(now), Json.set("lastGood", usage, Json.set("lastAttemptAt", Ustore.fl(now), release(row))))))))))
34
35def is_permanent(+e: String) -> Bool:
36  String.eq(e, "invalid_grant") || String.eq(e, "no_refresh_token")
37
38def with_strike(perm: Bool, +strikes: U32, struck: Json.Json, +row: Json.Json) -> Json.Json:
39  match perm:
40    case True{}:
41      Json.set("struckFingerprint", struck, Json.set("authDeadStrikes", Json.JNum{U32.show((strikes + 1 : U32))}, row))
42    case False{}:
43      row
44
45def with_429(rl: Bool, +now: U32, +row: Json.Json) -> Json.Json:
46  match rl:
47    case True{}:
48      Json.set("last429At", Ustore.fl(now), row)
49    case False{}:
50      row
51
52def apply_fail2(+row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +failures: U32, +strikes: U32, +now: U32) -> Json.Json:
53  with_strike(is_permanent(error), strikes, struck,
54    Json.set("backoffUntil", Ustore.fl((now + Pollrow.failure_backoff(failures, retry, String.eq(error, "http-429")) : U32)),
55      with_429(String.eq(error, "http-429"), now,
56        Json.set("lastError", Ustore.jstr(error), Json.set("consecutiveFailures", Json.JNum{U32.show(failures)},
57          Json.set("lastAttemptAt", Ustore.fl(now), release(row)))))))
58
59def apply_fail(+row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +now: U32) -> Json.Json:
60  apply_fail2(row, error, retry, struck, (Maybe.default(&2, U32, Pollrow.num.field("consecutiveFailures", row), 0) + 1 : U32),
61    Maybe.default(&2, U32, Pollrow.num.field("authDeadStrikes", row), 0), now)
62
63def apply_kind2(fail: Bool, +row: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, +now: U32) -> Json.Json:
64  match fail:
65    case True{}:
66      apply_fail(row, error, retry, struck, now)
67    case False{}:
68      release(row)
69
70def apply_kind.pick(ok: Bool, +kind: U32, +row: Json.Json, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
71  match ok:
72    case True{}:
73      apply_ok(row, usage, plan, now)
74    case False{}:
75      apply_kind2(U32.is_eq(kind, 1), row, error, retry, struck, now)
76
77def apply_kind(+kind: U32, +row: Json.Json, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, +now: U32) -> Json.Json:
78  apply_kind.pick(U32.is_eq(kind, 0), kind, row, usage, error, retry, struck, plan, now)
79
80type RA is Data:
81  RA{rows: Json.Json, done: List<&2, String>}
82
83def record.fenced(ok: Bool, +rows: Json.Json, +slot: String, +row: Json.Json, +kind: U32, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, done: List<&2, String>, +now: U32) -> RA:
84  match ok:
85    case True{}:
86      RA{Json.set(slot, apply_kind(kind, row, usage, error, retry, struck, plan, now), rows), slot <> done}
87    case False{}:
88      RA{rows, done}
89
90def record.apply(m: Maybe<&2, Json.Json>, +rows: Json.Json, +slot: String, +claim: String, +kind: U32, +usage: Json.Json, +error: String, retry: Maybe<&2, U32>, struck: Json.Json, plan: Maybe<&2, Poll.Sched>, done: List<&2, String>, +now: U32) -> RA:
91  match m:
92    case None{}:
93      RA{rows, done}
94    case Some{+row}:
95      record.fenced(String.eq(Maybe.default(&2, String, Json.as_string(Ustore.jv("claimId", row)), ""), claim), rows, slot, row, kind, usage, error, retry, struck, plan, done, now)
96
97def record.one(+rows: Json.Json, +r: Rec, done: List<&2, String>, +now: U32) -> RA:
98  match r:
99    case Rec{id, claim, kind, usage, error, retry, struck, plan}:
100      record.apply(Ustore.row_of(rows, id), rows, Ustore.slot_of(id), claim, kind, usage, error, retry, struck, plan, done, now)
101
102def record.go(rs: List<&2, Rec>, ra: RA, +now: U32) -> RA:
103  match rs ra:
104    case Nil{} _:
105      ra
106    case Con{h, t} RA{+rows, done}:
107      record.go(t, record.one(rows, h, done, now), now)
108
109def record.write(done: List<&2, String>, +root: String, +rows: Json.Json) -> Act.of(List<&2, String>):
110  match done:
111    case Nil{}:
112      Act.pure(List<&2, String>, [])
113    case Con{h, t}:
114      Act.bind(Unit, List<&2, String>, Ustore.write_rows(root, rows), u => Act.pure(List<&2, String>, h <> t))
115
116def record.commit(+root: String, ra: RA) -> Act.of(List<&2, String>):
117  match ra:
118    case RA{rows, done}:
119      record.write(done, root, rows)
120
121def record.act(+root: String, rs: List<&2, Rec>, +now: U32) -> Act.of(List<&2, String>):
122  Act.bind(Json.Json, List<&2, String>, Ustore.read_act(root), rows => record.commit(root, record.go(rs, RA{rows, []}, now)))
123
124# Merge outcomes fenced by the leases that produced them: a late writer whose
125# lease or identity was replaced is ignored. Success resets the failure
126# fields; a failure never touches lastGood and fetchedAt. Returns the slots
127# accepted.
128def record(+root: String, rs: List<&2, Rec>, +now: U32) -> Act.of(List<&2, String>):
129  Ustore.with_usage_lock(List<&2, String>, root, record.act(root, rs, now))
130
131type Reading is Data:
132  Reading{id: Ustore.Id, usage: Json.Json, age: U32}
133
134def hold.pick(later: Bool, +held: U32, +row: Json.Json) -> Json.Json:
135  match later:
136    case True{}:
137      Json.set("heldUntil", Ustore.fl(held), row)
138    case False{}:
139      Json.del("heldUntil", row)
140
141def hold.apply(+held: U32, +now: U32, +row: Json.Json) -> Json.Json:
142  hold.pick(U32.is_gt(held, now), held, row)
143
144def hold.reset(m: Maybe<&2, U32>, +held: U32) -> U32:
145  match m:
146    case Some{r}:
147      Poll.min_u(held, r)
148    case None{}:
149      held
150
151def hold.row(+h: U32, +stored: U32, +now: U32, +row: Json.Json) -> Json.Json:
152  hold.apply(hold.reset(Poll.earliest_reset(Poll.relevant(Pollrow.last_good.of(row), ["all"]), None{}), Poll.min_u((now + h : U32), (stored + Poll.trust_ceiling() : U32))), now, row)
153
154def adopt.hold(hold: Maybe<&2, U32>, +stored: U32, +now: U32, +row: Json.Json) -> Json.Json:
155  match hold:
156    case None{}:
157      row
158    case Some{h}:
159      hold.row(h, stored, now, row)
160
161def adopt.newer(replace: Bool, +usage: Json.Json, +at: U32, stored: Maybe<&2, U32>, hold: Maybe<&2, U32>, +now: U32, +row: Json.Json) -> Ustore.RAd:
162  match replace:
163    case True{}:
164      Ustore.RAd{adopt.hold(hold, at, now, Json.set("fetchedAt", Ustore.fl(at), Json.set("lastGood", usage, row))), True{}}
165    case False{}:
166      Ustore.RAd{adopt.hold(hold, Maybe.default(&2, U32, stored, at), now, row), False{}}
167
168def is_newer(stored: Maybe<&2, U32>, +at: U32) -> Bool:
169  match stored:
170    case None{}:
171      True{}
172    case Some{s}:
173      U32.is_gt(at, s)
174
175def adopt.row(+row: Json.Json, +usage: Json.Json, +age: U32, hold: Maybe<&2, U32>, +now: U32) -> Ustore.RAd:
176  adopt.newer(is_newer(Pollrow.num.field("fetchedAt", row), Poll.sub0(now, age)), usage, Poll.sub0(now, age), Pollrow.num.field("fetchedAt", row), hold, now, row)
177
178def adopt.stored(m: Maybe<&2, Json.Json>, +id: Ustore.Id) -> Json.Json:
179  match m:
180    case Some{row}:
181      row
182    case None{}:
183      Ustore.fresh_row(id)
184
185type AR is Data:
186  AR{rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>}
187
188def adopt.sort(yes: Bool, +slot: String, +rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>) -> AR:
189  match yes:
190    case True{}:
191      AR{rows, slot <> adopted, kept}
192    case False{}:
193      AR{rows, adopted, slot <> kept}
194
195def adopt.put(+rd: Ustore.RAd, +slot: String, +rows: Json.Json, adopted: List<&2, String>, kept: List<&2, String>) -> AR:
196  match rd:
197    case Ustore.RAd{row, yes}:
198      adopt.sort(yes, slot, Json.set(slot, row, rows), adopted, kept)
199
200def adopt.go(rs: List<&2, Reading>, ar: AR, +hold: Maybe<&2, U32>, +now: U32) -> AR:
201  match rs ar:
202    case Nil{} _:
203      ar
204    case Con{Reading{+id, usage, age}, t} AR{+rows, adopted, kept}:
205      adopt.go(t, adopt.put(adopt.row(adopt.stored(Ustore.row_of(rows, id), id), usage, age, hold, now), Ustore.slot_of(id), rows, adopted, kept), hold, now)
206
207def adopt.finish(+root: String, ar: AR) -> Act.of(AR):
208  match ar:
209    case AR{+rows, adopted, kept}:
210      Act.bind(Unit, AR, Ustore.write_rows(root, rows), u => Act.pure(AR, AR{rows, adopted, kept}))
211
212def adopt.act(+root: String, rs: List<&2, Reading>, hold: Maybe<&2, U32>, +now: U32) -> Act.of(AR):
213  Act.bind(Json.Json, AR, Ustore.read_act(root), rows => adopt.finish(root, adopt.go(rs, AR{rows, [], []}, hold, now)))
214
215# Merge measurements another machine took (UsageStore.adopt): a reading
216# replaces lastGood only when newer than the stored one; `hold` (seconds)
217# keeps collectors off the slot, never past the reading's earliest window
218# reset nor an hour past the stored measurement; Some{0} lifts a hold.
219def adopt(+root: String, rs: List<&2, Reading>, hold: Maybe<&2, U32>, +now: U32) -> Act.of(AR):
220  Ustore.with_usage_lock(AR, root, adopt.act(root, rs, hold, now))
221
222def cleared(+row: Json.Json) -> Json.Json:
223  Json.set("backoffUntil", Json.JNull{}, Json.set("lastError", Json.JNull{}, Json.set("consecutiveFailures", Json.JNum{"0"},
224    Json.set("struckFingerprint", Json.JNull{}, Json.set("authDeadStrikes", Json.JNum{"0"}, release(row))))))
225
226def clear.row(m: Maybe<&2, Json.Json>, +rows: Json.Json, +id: Ustore.Id) -> Json.Json:
227  match m:
228    case Some{row}:
229      Json.set(Ustore.slot_of(id), cleared(row), rows)
230    case None{}:
231      Json.set(Ustore.slot_of(id), cleared(Ustore.fresh_row(id)), rows)
232
233def clear.go(ids: List<&2, Ustore.Id>, +rows: Json.Json) -> Json.Json:
234  match ids:
235    case Nil{}:
236      rows
237    case Con{+id, t}:
238      clear.go(t, clear.row(Ustore.row_of(rows, id), rows, id))
239
240def clear.act(+root: String, ids: List<&2, Ustore.Id>) -> Act.of(Unit):
241  Act.bind(Json.Json, Unit, Ustore.read_act(root), rows => Ustore.write_rows(root, clear.go(ids, rows)))
242
243# Lift the dead-token quarantine of slots whose credential was replaced.
244def clear_dead(+root: String, ids: List<&2, Ustore.Id>) -> Act.of(Unit):
245  Ustore.with_usage_lock(Unit, root, clear.act(root, ids))
246