ustore.bend

cswap-bend-cli@0.1.0.0 · 211 lines · 40 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
9
10# API (also the daemon's usage store; do not copy it):
11#   read_rows(root) -> IO(Json)          the accounts object; empty for a missing/torn/other-version file
12#   row_of(rows, Id) -> Maybe<Json>      the row, only when it belongs to this (email, org) identity
13#   reserve(root, ids, now, respect, repair, tick) -> Act(List<Claim>)   win the right to fetch (90 s lease)
14#   record(root, recs, now) -> Act(List<slot>)   merge outcomes fenced by their lease (success, failure, sentinel)
15#   adopt(root, readings, hold, now) -> Act(AR)  merge another machine's readings, with an optional hold
16#   clear_dead(root, ids) -> Act(Unit)           reset a slot's row after its credential was replaced
17# Every write is read-modify-write under <root>/cache/.usage.lock; no network call runs under it.
18#
19# cache/usage.json (usage_store.py): rows keyed by slot, guarded by the
20# slot's (email, organizationUuid), read lock-free and written as
21# read-modify-write under cache/.usage.lock. No network call runs under the
22# lock: reserve() leases the rows, the caller fetches, record() merges what
23# the lease still covers.
24
25def ActM(-A: Type) -> Type:
26  Act.of(A)
27
28def ActM.pure(-A: Type, x: A) -> Act.of(A):
29  Act.pure(A, x)
30
31def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
32  Act.bind(A, B, m, f)
33
34def usage_path(+root: String) -> String:
35  root ++ "/cache/usage.json"
36
37def usage_lock(+root: String) -> String:
38  root ++ "/cache/.usage.lock"
39
40# The identity a slot number currently maps to.
41type Id is Data:
42  Id{slot: String, email: String, org: String}
43
44# A fetch lease: the slot and its fencing id.
45type Claim is Data:
46  Claim{slot: String, id: String}
47
48def jv(+k: String, +j: Json.Json) -> Json.Json:
49  Maybe.default(&2, Json.Json, Json.get(k, j), Json.JNull{})
50
51def jstr(s: String) -> Json.Json:
52  Json.JStr{s}
53
54# An epoch as the reference stores it: a float.
55def fl(+t: U32) -> Json.Json:
56  Json.JNum{U32.show(t) ++ ".0"}
57
58def empty_rows() -> Json.Json:
59  Json.of_pairs([])
60
61def obj_or_empty(j: Json.Json) -> Json.Json:
62  match j:
63    case Json.JObj{f}:
64      Json.JObj{f}
65    case _:
66      empty_rows()
67
68def rows.pick(ok: Bool, +j: Json.Json) -> Json.Json:
69  match ok:
70    case True{}:
71      obj_or_empty(jv("accounts", j))
72    case False{}:
73      empty_rows()
74
75def is_v2(j: Json.Json) -> Bool:
76  match j:
77    case Json.JNum{t}:
78      String.eq(t, "2")
79    case _:
80      False{}
81
82# The accounts object of a parsed usage.json; empty for any other schema.
83def rows_of(+j: Json.Json) -> Json.Json:
84  rows.pick(is_v2(jv("schemaVersion", j)), j)
85
86def rows.parsed(r: Result<&2, &2, String, Json.Json>) -> Json.Json:
87  match r:
88    case Fail{e}:
89      empty_rows()
90    case Done{j}:
91      rows_of(j)
92
93def rows.text(r: Fs.Res(String)) -> Json.Json:
94  match r:
95    case Fail{e}:
96      empty_rows()
97    case Done{t}:
98      rows.parsed(Json.parse(t))
99
100# A missing, torn or other-version file reads as empty.
101def read_rows(+root: String) -> IO(Json.Json):
102  IO.bind(Fs.Res(String), Json.Json, Fs.read_file(usage_path(root)), r => IO.pure(Json.Json, rows.text(r)))
103
104def matches(+row: Json.Json, +id: Id) -> Bool:
105  match id:
106    case Id{slot, email, org}:
107      String.eq(Maybe.default(&2, String, Json.as_string(jv("email", row)), "\u{1}"), email)
108        && String.eq(Maybe.default(&2, String, Json.as_string(Maybe.default(&2, Json.Json, Json.get("organizationUuid", row), Json.JStr{""})), "\u{1}"), org)
109
110def fresh_row(+id: Id) -> Json.Json:
111  match id:
112    case Id{slot, email, org}:
113      Json.of_pairs([("email", jstr(email)), ("organizationUuid", jstr(org))])
114
115def row.if(ok: Bool, +row: Json.Json) -> Maybe<&2, Json.Json>:
116  match ok:
117    case True{}:
118      Some{row}
119    case False{}:
120      None{}
121
122def row.pick(m: Maybe<&2, Json.Json>, +id: Id) -> Maybe<&2, Json.Json>:
123  match m:
124    case None{}:
125      None{}
126    case Some{+row}:
127      row.if(matches(row, id), row)
128
129# The stored row of a slot, only when it belongs to this identity.
130def row_of(+rows: Json.Json, +id: Id) -> Maybe<&2, Json.Json>:
131  match id:
132    case Id{slot, email, org}:
133      row.pick(Json.get(slot, rows), id)
134
135def write_rows(+root: String, +rows: Json.Json) -> Act.of(Unit):
136  do ActM<Unit>:
137    FWrite.mkdir_p(root ++ "/cache")
138    FWrite.write_atomic(usage_path(root), Json.emit_pretty(Json.of_pairs([("schemaVersion", Json.JNum{"2"}), ("accounts", rows)])), "600")
139
140def with_usage_lock(-A: Type, +root: String, act: Act.of(A)) -> Act.of(A):
141  Lock.with_lock(A, usage_lock(root), 10000, act)
142
143def read_act(+root: String) -> Act.of(Json.Json):
144  Act.lift(Json.Json, read_rows(root))
145
146type RR is Data:
147  RR{rows: Json.Json, won: List<&2, Claim>}
148
149def claim_row(+row: Json.Json, +claim: String, +now: U32) -> Json.Json:
150  Json.set("claimUntil", fl((now + Poll.claim_ttl() : U32)), Json.set("claimId", jstr(claim), Json.set("lastAttemptAt", fl(now), row)))
151
152def reserve.take(+rows: Json.Json, +id: Id, +row: Json.Json, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
153  match id:
154    case Id{slot, email, org}:
155      RR{Json.set(slot, claim_row(row, claim, now), rows), Claim{slot, claim} <> won}
156
157def reserve.fresh(+rows: Json.Json, +id: Id, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
158  reserve.take(rows, id, fresh_row(id), claim, now, won)
159
160def reserve.stored(ok: Bool, +rows: Json.Json, +id: Id, +row: Json.Json, +claim: String, +now: U32, won: List<&2, Claim>) -> RR:
161  match ok:
162    case True{}:
163      reserve.take(rows, id, row, claim, now, won)
164    case False{}:
165      RR{rows, won}
166
167def reserve.row(m: Maybe<&2, Json.Json>, +rows: Json.Json, +id: Id, +claim: String, +now: U32, +respect: Bool, +repair: Bool, won: List<&2, Claim>) -> RR:
168  match m:
169    case None{}:
170      reserve.fresh(rows, id, claim, now, won)
171    case Some{+row}:
172      reserve.stored(Pollrow.row_eligible(Pollrow.row.of(row), now, respect, repair), rows, id, row, claim, now, won)
173
174def claim_id(+slot: String, +now: U32, +tick: Nat) -> String:
175  "c" ++ U32.show(now) ++ "-" ++ slot ++ "-" ++ Nat.show(tick)
176
177def slot_of(+id: Id) -> String:
178  match id:
179    case Id{slot, email, org}:
180      slot
181
182def reserve.go(ids: List<&2, Id>, rr: RR, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> RR:
183  match ids rr:
184    case Nil{} _:
185      rr
186    case Con{+id, t} RR{+rows, won}:
187      reserve.go(t, reserve.row(row_of(rows, id), rows, id, claim_id(slot_of(id), now, tick), now, respect, repair, won), now, respect, repair, 1n+tick)
188
189def reserve.write(won: List<&2, Claim>, +root: String, +rows: Json.Json) -> Act.of(List<&2, Claim>):
190  match won:
191    case Nil{}:
192      Act.pure(List<&2, Claim>, [])
193    case Con{h, t}:
194      Act.bind(Unit, List<&2, Claim>, write_rows(root, rows), u => Act.pure(List<&2, Claim>, h <> t))
195
196def reserve.commit(+root: String, rr: RR) -> Act.of(List<&2, Claim>):
197  match rr:
198    case RR{rows, won}:
199      reserve.write(won, root, rows)
200
201def reserve.act(+root: String, ids: List<&2, Id>, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> Act.of(List<&2, Claim>):
202  Act.bind(Json.Json, List<&2, Claim>, read_act(root), rows => reserve.commit(root, reserve.go(ids, RR{rows, []}, now, respect, repair, tick)))
203
204# Atomically win the right to fetch: re-check each slot's eligibility and
205# stamp a 90 s lease under the lock (UsageStore.reserve). `respect` is the
206# on-demand mode (stale AND poll-due), `repair` also frees a reset-parked plan.
207def reserve(+root: String, ids: List<&2, Id>, +now: U32, +respect: Bool, +repair: Bool, +tick: Nat) -> Act.of(List<&2, Claim>):
208  with_usage_lock(List<&2, Claim>, root, reserve.act(root, ids, now, respect, repair, tick))
209
210type RAd is Data:
211  RAd{row: Json.Json, adopted: Bool}