daemon/auto.bend

cswap-bend-cli@0.1.0.0 · 356 lines · 61 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/json.bend as Json
3import cswap-bend-cli@0.1.0.0/fs.bend as Fs
4import cswap-bend-cli@0.1.0.0/fs/write.bend as FsW
5import cswap-bend-cli@0.1.0.0/act.bend as Act
6import cswap-bend-cli@0.1.0.0/clock.bend as Clock
7import cswap-bend-cli@0.1.0.0/store.bend as Store
8import cswap-bend-cli@0.1.0.0/store/base.bend as SBase
9import cswap-bend-cli@0.1.0.0/swlive.bend as L
10import cswap-bend-cli@0.1.0.0/policy/types.bend as T
11import cswap-bend-cli@0.1.0.0/daemon/events.bend as Ev
12import cswap-bend-cli@0.1.0.0/fmt/out.bend as O
13import cswap-bend-cli@0.1.0.0/daemon/cadence.bend as Cad
14import cswap-bend-cli@0.1.0.0/daemon/config.bend as Cfg
15import cswap-bend-cli@0.1.0.0/daemon/live.bend as Lv
16import cswap-bend-cli@0.1.0.0/store/io.bend as SIo
17import cswap-bend-cli@0.1.0.0/daemon/walk.bend as Wk
18import cswap-bend-cli@0.1.0.0/daemon/quarantine.bend as Q
19import cswap-bend-cli@0.1.0.0/daemon/dec.bend as Dc
20import cswap-bend-cli@0.1.0.0/daemon/tick.bend as Tk
21import cswap-bend-cli@0.1.0.0/daemon/gather.bend as G
22import cswap-bend-cli@0.1.0.0/daemon/collect.bend as Col
23import cswap-bend-cli@0.1.0.0/daemon/status.bend as Stat
24import cswap-bend-cli@0.1.0.0/daemon/pidlock.bend as Pid
25
26# `cswap auto`: --once is one tick (exit 0 switched, 1 error, 2 no action, 3
27# blocked); otherwise tick, sleep, tick, forever. One decider per backup root
28# (a pid-file flock), the status file rewritten after every tick, and a JSONL
29# log line per tick with where each account's reading came from. A SIGTERM
30# ends the process (the kernel drops the lock); the pid file is then stale
31# but harmless, as the lock is what counts.
32
33# What a run is: the roster context, the state dir, the settings in force,
34# --json, --dry-run, the switch options, and when the process started, the
35# pid, and what is needed to read the settings again each tick (the flags and
36# the profile in force: src/daemon/live.bend).
37
38type Run is Data:
39  Run{ctx: SBase.Ctx, state_dir: String, cfg: Cfg.Cfg, fmt: O.Fmt, dry: Bool, opts: L.Opts, started: String, version: String, pid: U32, live: Lv.Live}
40
41def ActM(-A: Type) -> Type:
42  Act.of(A)
43
44def ActM.pure(-A: Type, x: A) -> Act.of(A):
45  Act.pure(A, x)
46
47def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
48  Act.bind(A, B, m, f)
49
50def run_ctx(r: Run) -> SBase.Ctx:
51  match r:
52    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
53      c
54
55def run_dir(r: Run) -> String:
56  match r:
57    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
58      sd
59
60def run_cfg(r: Run) -> Cfg.Cfg:
61  match r:
62    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
63      cf
64
65def run_fmt(r: Run) -> O.Fmt:
66  match r:
67    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
68      j
69
70def run_dry(r: Run) -> Bool:
71  match r:
72    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
73      d
74
75def run_opts(r: Run) -> L.Opts:
76  match r:
77    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
78      o
79
80def run_started(r: Run) -> String:
81  match r:
82    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
83      st
84
85def run_version(r: Run) -> String:
86  match r:
87    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
88      v
89
90def run_pid(r: Run) -> U32:
91  match r:
92    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
93      p
94
95def run_live(r: Run) -> Lv.Live:
96  match r:
97    case Run{c, sd, cf, j, d, o, st, v, p, lv}:
98      lv
99
100def with_live(+r: Run, +cfg: Cfg.Cfg, +lv: Lv.Live) -> Run:
101  match r:
102    case Run{c, sd, cf, j, d, o, st, v, p, l0}:
103      Run{c, sd, cfg, j, d, o, st, v, p, lv}
104
105# The run with the settings of this tick: settings.json read again, the
106# profile laid over it, the flags over both.
107def refreshed(+r: Run, +text: String) -> Run:
108  with_live(r, Lv.cfg_for(text, Lv.live_flags(run_live(r)), Lv.live_over(run_live(r))), Lv.live_for(text, Lv.live_flags(run_live(r)), Lv.live_over(run_live(r))))
109
110def refresh(+r: Run) -> Act.of(Run):
111  do ActM<Run>:
112    text : String <- Act.lift(String, Lv.read_text(SBase.root_of(run_ctx(r))))
113    return refreshed(r, text)
114
115def tagged(+r: Run, +o: Tk.Out) -> Tk.Out:
116  match o:
117    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
118      Tk.Out{Dc.TickRes{i, Lv.tag_events(Lv.live_tag(run_live(r)), evs)}, m, a, es, ac, ck}
119
120def interval(r: Run) -> U32:
121  Cfg.cfg_interval(run_cfg(r))
122
123# ---- printing --------------------------------------------------------------------------------------------
124
125def show2(human: Bool, +f: O.Fmt, +e: Json.Json) -> String:
126  match human:
127    case True{}:
128      Ev.human(e)
129    case False{}:
130      O.event_text(f, e)
131
132def show(+f: O.Fmt, +e: Json.Json) -> String:
133  show2(O.is_text(f), f, e)
134
135def print_all(+f: O.Fmt, +evs: List<&2, Json.Json>) -> Act.of(Unit):
136  match evs:
137    case Nil{}:
138      Act.pure(Unit, Unit{})
139    case Con{e, t}:
140      do ActM<Unit>:
141        Act.lift(Unit, IO.print(show(f, e)))
142        print_all(f, t)
143
144# ---- one tick, never raising ------------------------------------------------------------------------------
145
146def failed_out(+env: Tk.Env, +ts: String, +msg: String) -> Tk.Out:
147  Tk.Out{Dc.TickRes{Cad.Info{Cad.Failed{}, None{}, False{}, False{}, None{}}, [Ev.error(ts, msg, True{})]}, Tk.env_mem(env), None{}, [], Json.arr([]), Tk.env_checked(env)}
148
149def prepend(+extra: List<&2, Json.Json>, +o: Tk.Out) -> Tk.Out:
150  match o:
151    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
152      Tk.Out{Dc.TickRes{i, List.append(&2, Json.Json, extra, evs)}, m, a, es, ac, ck}
153
154def tick_out(+env: Tk.Env, +ts: String, r: Fs.Res(Tk.Out)) -> Tk.Out:
155  match r:
156    case Done{o}:
157      o
158    case Fail{(code, msg)}:
159      failed_out(env, ts, msg)
160
161def release_with(+env: Tk.Env, +t: U32, +sq: Maybe<&2, Json.Json>) -> Act.of(List<&2, Json.Json>):
162  match sq:
163    case None{}:
164      Act.pure(List<&2, Json.Json>, [])
165    case Some{seq}:
166      Q.release_recovered(SBase.root_of(Tk.env_ctx(env)), seq, Clock.iso8601_utc(t))
167
168def release_now(+env: Tk.Env, +t: U32) -> Act.of(List<&2, Json.Json>):
169  do ActM<List<&2, Json.Json>>:
170    sq : Maybe<&2, Json.Json> <- SIo.read_seq(SBase.root_of(Tk.env_ctx(env)))
171    release_with(env, t, sq)
172
173# Quarantines that no longer hold (the account was replaced, or its login was
174# renewed) are released before deciding; a dry run changes nothing.
175def released(+env: Tk.Env, +t: U32) -> Act.of(List<&2, Json.Json>):
176  Act.cond(List<&2, Json.Json>, Tk.env_dry(env), Act.pure(List<&2, Json.Json>, []), release_now(env, t))
177
178def fetched(+env: Tk.Env, +t: U32) -> Act.of(Maybe<&2, G.Snap>):
179  do ActM<Maybe<&2, G.Snap>>:
180    m0 : Maybe<&2, G.Snap> <- G.snapshot(Tk.env_ctx(env))
181    Col.refresh_opt(Tk.env_ctx(env), Tk.env_cfg(env), t, m0)
182
183# Release what no longer holds, snapshot, fetch what is due, decide, and walk the ranking.
184def look(+env: Tk.Env) -> Act.of(Tk.Out):
185  do ActM<Tk.Out>:
186    +t : U32 <- Act.lift(U32, Clock.now_s())
187    rel : List<&2, Json.Json> <- released(env, t)
188    m : Maybe<&2, G.Snap> <- fetched(env, t)
189    o : Tk.Out <- Wk.run(env, t, Clock.iso8601_utc(t), m)
190    return prepend(rel, o)
191
192def tick_safe(+env: Tk.Env) -> Act.of(Tk.Out):
193  do ActM<Tk.Out>:
194    t : U32 <- Act.lift(U32, Clock.now_s())
195    r : Fs.Res(Tk.Out) <- Act.lift(Fs.Res(Tk.Out), look(env))
196    return tick_out(env, Clock.iso8601_utc(t), r)
197
198# ---- after a tick: status file and log --------------------------------------------------------------------
199
200def out_info(+o: Tk.Out) -> Cad.Info:
201  match o:
202    case Tk.Out{Dc.TickRes{info, evs}, m, a, es, ac, ck}:
203      info
204
205def out_events(+o: Tk.Out) -> List<&2, Json.Json>:
206  match o:
207    case Tk.Out{Dc.TickRes{info, evs}, m, a, es, ac, ck}:
208      evs
209
210def out_mem(+o: Tk.Out) -> T.Mem:
211  match o:
212    case Tk.Out{r, m, a, es, ac, ck}:
213      m
214
215def out_active(+o: Tk.Out) -> Maybe<&2, String>:
216  match o:
217    case Tk.Out{r, m, a, es, ac, ck}:
218      a
219
220def out_entries(+o: Tk.Out) -> List<&2, Stat.Entry>:
221  match o:
222    case Tk.Out{r, m, a, es, ac, ck}:
223      es
224
225def out_checked(+o: Tk.Out) -> Bool:
226  match o:
227    case Tk.Out{r, m, a, es, ac, ck}:
228      ck
229
230def out_accts(+o: Tk.Out) -> Json.Json:
231  match o:
232    case Tk.Out{r, m, a, es, ac, ck}:
233      ac
234
235def info_outcome(i: Cad.Info) -> Cad.Outcome:
236  match i:
237    case Cad.Info{o, u, l, idle, np}:
238      o
239
240def outcome_name(o: Cad.Outcome) -> String:
241  match o:
242    case Cad.Switched{}:
243      "switched"
244    case Cad.NoAction{}:
245      "no-action"
246    case Cad.Blocked{}:
247      "blocked"
248    case Cad.Failed{}:
249      "failed"
250
251def status_of(+r: Run, +o: Tk.Out, +ts: String, +next_at: String) -> Stat.Status:
252  Stat.Status{run_pid(r), run_version(r), run_started(r), Some{ts}, Some{next_at}, out_active(o), out_entries(o), run_cfg(r)}
253
254# A blocked tick is a warning, a failed one an error.
255def level_of(o: Cad.Outcome) -> String:
256  match o:
257    case Cad.Blocked{}:
258      "warn"
259    case Cad.Failed{}:
260      "error"
261    case _:
262      "info"
263
264# One JSONL log line: the tick, what came of it, and per account where its
265# reading came from (usage-endpoint, header-probe or cache).
266def log_json(+ts: String, +o: Tk.Out) -> Json.Json:
267  Json.JObj{Json.JField{"ts", Json.JStr{ts}, Json.JField{"level", Json.JStr{level_of(info_outcome(out_info(o)))}, Json.JField{"msg", Json.JStr{"tick"},
268    Json.JField{"outcome", Json.JStr{outcome_name(info_outcome(out_info(o)))}, Json.JField{"active", Ev.opt_str(out_active(o)), Json.JField{"accounts", out_accts(o), Json.JNil{}}}}}}}}
269
270def swallow(r: Fs.Res(Unit)) -> Act.of(Unit):
271  Act.pure(Unit, Unit{})
272
273def best_effort(m: Act.of(Unit)) -> Act.of(Unit):
274  IO.bind(Fs.Res(Unit), Fs.Res(Unit), m, r => swallow(r))
275
276def write_status(+r: Run, +o: Tk.Out, +ts: String, +delay: U32, +now: U32) -> Act.of(Unit):
277  best_effort(Stat.write(run_dir(r), status_of(r, o, ts, Clock.iso8601_utc((now + delay : U32)))))
278
279def write_log(+r: Run, +o: Tk.Out, +ts: String) -> Act.of(Unit):
280  best_effort(FsW.append(run_dir(r) ++ "/daemon.log", Json.emit_py(Lv.tag_log(Lv.live_tag(run_live(r)), log_json(ts, o))) ++ "\n"))
281
282def record_tick(+r: Run, +o: Tk.Out, +ts: String, +delay: U32, +now: U32) -> Act.of(Unit):
283  do ActM<Unit>:
284    write_status(r, o, ts, delay, now)
285    write_log(r, o, ts)
286
287def sleep_if(+c: Bool, +e: Json.Json) -> List<&2, Json.Json>:
288  match c:
289    case True{}:
290      [e]
291    case False{}:
292      []
293
294def sleep_event(r: Run, +delay: U32, +now: U32) -> List<&2, Json.Json>:
295  sleep_if(Cad.announce(delay, interval(r)), Ev.sleep(Clock.iso8601_utc(now), (delay * 10 : U32), Clock.iso8601_utc((now + delay : U32))))
296
297def env_of(+r: Run, +mem: T.Mem, +checked: Bool) -> Tk.Env:
298  Tk.Env{run_ctx(r), run_cfg(r), run_dry(r), run_opts(r), mem, checked, []}
299
300# ---- --once --------------------------------------------------------------------------------------------------
301
302def finish_tick(+r: Run, +o: Tk.Out, +delay: U32, +now: U32, +extra: List<&2, Json.Json>) -> Act.of(Unit):
303  do ActM<Unit>:
304    print_all(run_fmt(r), List.append(&2, Json.Json, out_events(o), extra))
305    record_tick(r, o, Clock.iso8601_utc(now), delay, now)
306
307def first_tick(+r: Run) -> Act.of(Tk.Out):
308  do ActM<Tk.Out>:
309    +o : Tk.Out <- tick_safe(env_of(r, T.fresh_mem(), False{}))
310    return tagged(r, o)
311
312def out_code(+o: Tk.Out) -> U32:
313  Cad.exit_code(info_outcome(out_info(o)))
314
315def once(+r: Run) -> Act.of(U32):
316  do ActM<U32>:
317    +r2 : Run <- refresh(r)
318    +o : Tk.Out <- first_tick(r2)
319    now : U32 <- Act.lift(U32, Clock.now_s())
320    finish_tick(r2, o, interval(r2), now, [])
321    return out_code(o)
322
323# ---- the loop ------------------------------------------------------------------------------------------------
324
325# Tick, record, sleep, repeat. Not terminating by construction: kept in this
326# module, which no proof imports.
327def nap(+delay: U32) -> Act.of(Unit):
328  Act.lift(Unit, IO.sleep((delay * 1000 : U32)))
329
330def pace(+r: Run, +o: Tk.Out, +now: U32) -> Act.of(Unit):
331  do ActM<Unit>:
332    +delay : U32 = Cad.next_delay(out_info(o), now, interval(r))
333    finish_tick(r, o, delay, now, sleep_event(r, delay, now))
334    nap(delay)
335
336def settle(r: Run, +o: Tk.Out) -> Act.of(Unit):
337  do ActM<Unit>:
338    now : U32 <- Act.lift(U32, Clock.now_s())
339    pace(r, o, now)
340
341def tick_tagged(+r: Run, +mem: T.Mem, +checked: Bool) -> Act.of(Tk.Out):
342  do ActM<Tk.Out>:
343    +o : Tk.Out <- tick_safe(env_of(r, mem, checked))
344    return tagged(r, o)
345
346def round(+r: Run, +mem: T.Mem, +checked: Bool) -> Act.of(Tk.Out):
347  do ActM<Tk.Out>:
348    +r2 : Run <- refresh(r)
349    +o : Tk.Out <- tick_tagged(r2, mem, checked)
350    settle(r2, o)
351    return o
352
353@unsafe def forever(+r: Run, mem: T.Mem, checked: Bool) -> Act.of(U32):
354  do ActM<U32>:
355    +o : Tk.Out <- round(r, mem, checked)
356    forever(r, out_mem(o), out_checked(o))