daemon/shadow.bend

cswap-bend-cli@0.1.0.0 · 194 lines · 35 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/base.bend as SBase
8import cswap-bend-cli@0.1.0.0/swlive.bend as L
9import cswap-bend-cli@0.1.0.0/policy/types.bend as T
10import cswap-bend-cli@0.1.0.0/daemon/events.bend as Ev
11import cswap-bend-cli@0.1.0.0/daemon/config.bend as Cfg
12import cswap-bend-cli@0.1.0.0/daemon/dec.bend as Dc
13import cswap-bend-cli@0.1.0.0/daemon/tick.bend as Tk
14import cswap-bend-cli@0.1.0.0/daemon/gather.bend as G
15import cswap-bend-cli@0.1.0.0/daemon/pidlock.bend as Pid
16
17# `cswap auto --shadow --shadow-log PATH`: the decider run for comparison only.
18# It reads the usage cache the other poller keeps (no HTTP, no fetch), decides
19# with the same policy, and appends what it WOULD have done as one JSONL line
20# per tick to PATH. It never switches, quarantines, releases or touches state,
21# status, cache or log under the backup root; its only writes are PATH and a
22# pid/lock pair beside it.
23
24# The headroom (slot -> percent) the tick used: the poll event's headroomPct.
25
26type Step is Data:
27  Step{line: Json.Json, mem: T.Mem, checked: Bool}
28
29type Sh is Data:
30  Sh{ctx: SBase.Ctx, cfg: Cfg.Cfg, opts: L.Opts, log: String}
31
32def field2(+m: Maybe<&2, Json.Json>) -> Json.Json:
33  match m:
34    case None{}:
35      Json.JNull{}
36    case Some{j}:
37      j
38
39def field(+k: String, +e: Json.Json) -> Json.Json:
40  field2(Json.get(k, e))
41
42def headroom_of(+evs: List<&2, Json.Json>) -> Json.Json:
43  match evs:
44    case Nil{}:
45      Json.JObj{Json.JNil{}}
46    case Con{e, t}:
47      field("headroomPct", e)
48
49def ActM(-A: Type) -> Type:
50  Act.of(A)
51
52def ActM.pure(-A: Type, x: A) -> Act.of(A):
53  Act.pure(A, x)
54
55def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
56  Act.bind(A, B, m, f)
57
58def last_event(+es: List<&2, Json.Json>, +d: Json.Json) -> Json.Json:
59  match es:
60    case Nil{}:
61      d
62    case Con{e, t}:
63      last_event(t, e)
64
65def ctx_active(+c: Dc.Ctx) -> Json.Json:
66  match c:
67    case Dc.Ctx{x, d, cf, n, ts, o, a, np}:
68      a
69
70def out_events(+o: Tk.Out) -> List<&2, Json.Json>:
71  match o:
72    case Tk.Out{Dc.TickRes{i, evs}, m, a, es, ac, ck}:
73      evs
74
75def out_active(+o: Tk.Out) -> Maybe<&2, String>:
76  match o:
77    case Tk.Out{r, m, a, es, ac, ck}:
78      a
79
80def out_accts(+o: Tk.Out) -> Json.Json:
81  match o:
82    case Tk.Out{r, m, a, es, ac, ck}:
83      ac
84
85def out_mem(+o: Tk.Out) -> T.Mem:
86  match o:
87    case Tk.Out{r, m, a, es, ac, ck}:
88      m
89
90def out_checked(+o: Tk.Out) -> Bool:
91  match o:
92    case Tk.Out{r, m, a, es, ac, ck}:
93      ck
94
95# The decision as the event the live daemon would have logged.
96def decision_event(+ts: String, +c: Dc.Ctx, +d: Dc.Dec) -> Json.Json:
97  match d:
98    case Dc.DNone{reason, detail}:
99      Ev.no_switch(ts, reason, detail)
100    case Dc.DExhausted{e}:
101      Ev.no_switch(ts, "all-exhausted", "")
102    case Dc.DGo{to, email, trig, lh, lr}:
103      Ev.switch(ts, trig, ctx_active(c), Ev.ref(to, email), [], True{})
104
105def line(+ts: String, +o: Tk.Out, +dec: Json.Json, +evs: List<&2, Json.Json>) -> Json.Json:
106  Json.JObj{Json.JField{"ts", Json.JStr{ts}, Json.JField{"active", Ev.opt_str(out_active(o)),
107    Json.JField{"event", field("event", dec), Json.JField{"reason", field("reason", dec), Json.JField{"trigger", field("trigger", dec),
108    Json.JField{"from", field("from", dec), Json.JField{"to", field("to", dec), Json.JField{"accounts", headroom_of(evs),
109    Json.JField{"sources", out_accts(o), Json.JField{"events", Json.arr(evs), Json.JNil{}}}}}}}}}}}}
110
111def step_of(+ts: String, +o: Tk.Out, +dec: Json.Json, +evs: List<&2, Json.Json>) -> Step:
112  Step{line(ts, o, dec, evs), out_mem(o), out_checked(o)}
113
114def shadow2(+ts: String, +nx: Tk.Next) -> Step:
115  match nx:
116    case Tk.NDone{o}:
117      step_of(ts, o, last_event(out_events(o), Json.JNull{}), out_events(o))
118    case Tk.NAct{o, c, d}:
119      step_of(ts, o, decision_event(ts, c, d), List.append(&2, Json.Json, out_events(o), [decision_event(ts, c, d)]))
120
121# What this tick would do: the finished outcome's own last event, or the
122# decision it stopped at. Pure: nothing is fetched, applied or written.
123def shadow_step(+env: Tk.Env, +t: U32, +m: Maybe<&2, G.Snap>) -> Step:
124  shadow2(Clock.iso8601_utc(t), Tk.tick(env, t, m))
125
126def step_line(+s: Step) -> String:
127  match s:
128    case Step{l, m, c}:
129      Json.emit_py(l) ++ "\n"
130
131def step_mem(+s: Step) -> T.Mem:
132  match s:
133    case Step{l, m, c}:
134      m
135
136def step_checked(+s: Step) -> Bool:
137  match s:
138    case Step{l, m, c}:
139      c
140
141def swallow(r: Fs.Res(Unit)) -> Act.of(Unit):
142  Act.pure(Unit, Unit{})
143
144# ---- the loop ------------------------------------------------------------------------------------------------
145
146def sh_env(+r: Sh, +mem: T.Mem, +checked: Bool) -> Tk.Env:
147  match r:
148    case Sh{c, cf, o, l}:
149      Tk.Env{c, cf, True{}, o, mem, checked, []}
150
151def sh_ctx(+r: Sh) -> SBase.Ctx:
152  match r:
153    case Sh{c, cf, o, l}:
154      c
155
156def sh_log(+r: Sh) -> String:
157  match r:
158    case Sh{c, cf, o, l}:
159      l
160
161def sh_interval(+r: Sh) -> U32:
162  match r:
163    case Sh{c, cf, o, l}:
164      Cfg.cfg_interval(cf)
165
166def emit(+r: Sh, +s: Step) -> Act.of(Step):
167  IO.bind(Fs.Res(Unit), Fs.Res(Step), FsW.append(sh_log(r), step_line(s)), u => IO.pure(Fs.Res(Step), Done{s}))
168
169def one_tick(+r: Sh, +mem: T.Mem, +checked: Bool) -> Act.of(Step):
170  do ActM<Step>:
171    +t : U32 <- Act.lift(U32, Clock.now_s())
172    m : Maybe<&2, G.Snap> <- G.snapshot(sh_ctx(r))
173    emit(r, shadow_step(sh_env(r, mem, checked), t, m))
174
175def once(+r: Sh) -> Act.of(U32):
176  do ActM<U32>:
177    s : Step <- one_tick(r, T.fresh_mem(), False{})
178    return 0
179
180def nap(+delay: U32) -> Act.of(Unit):
181  Act.lift(Unit, IO.sleep((delay * 1000 : U32)))
182
183@unsafe def forever(+r: Sh, mem: T.Mem, checked: Bool) -> Act.of(U32):
184  do ActM<U32>:
185    +s : Step <- one_tick(r, mem, checked)
186    nap(sh_interval(r))
187    forever(r, step_mem(s), step_checked(s))
188
189def body(+r: Sh, +once_only: Bool) -> Act.of(U32):
190  Act.cond(U32, once_only, once(r), forever(r, T.fresh_mem(), False{}))
191
192# Run as the one shadow decider of the log's directory.
193def run(+r: Sh, +once_only: Bool) -> Act.of(U32):
194  Pid.with_pidlock(U32, Fs.dirname(sh_log(r)), body(r, once_only))