src/runtime.bend

bend-kube-runtime@0.2.0.0 · 719 lines · 93 definitions

1import Base
2import bend-kube-runtime@0.2.0.0/src/json.bend as J
3import bend-kube-runtime@0.2.0.0/src/utf8.bend as U
4import bend-kube-runtime@0.2.0.0/src/http.bend as H
5import bend-kube-runtime@0.2.0.0/src/core.bend as C
6import bend-kube-runtime@0.2.0.0/src/config.bend as K
7import bend-kube-runtime@0.2.0.0/src/net.bend as N
8import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
9import bend-kube-runtime@0.2.0.0/src/watcher.bend as W
10import bend-kube-runtime@0.2.0.0/src/store.bend as S
11import bend-kube-runtime@0.2.0.0/src/queue.bend as Q
12import bend-kube-runtime@0.2.0.0/src/log.bend as L
13
14# Runtime
15# =======
16#
17# kube-runtime's Controller in Bend. Watchers run as their own IO
18# computations and send what they see to one channel; the controller loop
19# owns the store and the queue (both pure values), starts reconciles as
20# concurrent computations, and arms timers, all of which answer on the
21# same channel. Every decision is made by the pure modules (watcher.bend,
22# store.bend, queue.bend); this file only performs them.
23#
24# Loops count down a large fuel (4294967295n steps, lazily), so they
25# terminate by structure; at a few steps a second that is over a year.
26
27type Msg is Data:
28  MOuts{src: U32, outs: List<&2, W.Out>}
29  MDone{key: String, act: Q.Action}
30  MWake{key: String, gen: U32}
31
32# The controller's state as the health server reports it.
33type Snap is Data:
34  Snap{name: String, ready: Bool, reconciles: U32, errors: U32, events: U32, running: U32,
35    objects: U32}
36
37# The status keeper's inbox: a controller's latest snapshot (by its index),
38# or a query for all of them (none until a controller runs: a standby
39# replica).
40type SMsg is Data:
41  SPut{idx: U32, snap: Snap}
42  SGet{reply: Chan(List<&2, Snap>)}
43
44# Counters since start.
45type Stats is Data:
46  Stats{reconciles: U32, errors: U32, events: U32}
47
48# Watcher driver
49# --------------
50
51# Who a watcher feeds: each subscriber's inbox, and the source number the
52# outputs carry there (0 for a controller's primary, 1.. for its owned).
53type Sub is Data:
54  Sub{chan: Chan(Msg), src: U32}
55
56type WCtx is Data:
57  WCtx{cfg: K.Config, api: C.Api, label: String, field: String, subs: List<&2, Sub>}
58
59# What the driver just saw.
60type DEv is Type:
61  DStart{}
62  DListed{r: Result<&2, &2, Cl.KErr, J.Json>}
63  DOpened{r: Result<&1, &1, U32 & String, N.Tls>}
64  DRecv{sock: Socket, r: Result<&1, &1, U32 & String, List<&2, U32>>}
65  DDig{sock: Socket, dig: Dig}
66  DStepped{step: W.Step, sock: Maybe<&1, Socket>, more: Bool}
67  DNext{sock: Maybe<&1, Socket>, wait: Bool, more: Bool}
68
69def DEv.recv(p: Socket & Result<&1, &1, U32 & String, List<&2, U32>>) -> DEv:
70  match p:
71    case (s, r):
72      DRecv{s, r}
73
74# The inputs a run of watch lines makes; blank lines are skipped.
75def Drv.inputs(ls: List<&2, List<&2, U32>>) -> List<&2, W.Input>:
76  match ls:
77    case Nil{}:
78      Nil{}
79    case Con{Nil{}, t}:
80      Drv.inputs(t)
81    case Con{l, t}:
82      W.IEvent{C.WEvent.parse(U.Utf8.decode_all(l))} <> Drv.inputs(t)
83
84# The same, with the lines decoded and parsed in parallel: each line is
85# its own JSON document, so the halves split (to depth levels, up to 2^depth
86# tasks) run as Bend parallel calls, on every core.
87def Drv.inputs.par(+depth: Nat, +ls: List<&2, List<&2, U32>>) -> List<&2, W.Input>:
88  match depth:
89    case 0n:
90      Drv.inputs(ls)
91    case 1n+d:
92      +half = Nat.div(List.length(&2, List<&2, U32>, ls), 2n)
93      a b = Drv.inputs.par(d, List.take(&2, List<&2, U32>, ls, half)) Drv.inputs.par(d, List.drop(&2, List<&2, U32>, ls, half))
94      List.append(&2, W.Input, a, b)
95
96# After a recv: the stream's state, and the step its whole lines make.
97type Dig is Data:
98  Dig{step: W.Step, resp: H.Resp, carry: List<&2, U32>, more: Bool}
99
100def Drv.status(r: H.Resp) -> U32:
101  match r:
102    case H.Resp{p, Some{H.Head{c, reason, hs}}, o}:
103      c
104    case _:
105      200
106
107def Drv.ended(r: H.Resp) -> Bool:
108  match r:
109    case H.Resp{H.PDone{}, h, o}:
110      True{}
111    case H.Resp{H.PErr{m}, h, o}:
112      True{}
113    case H.Resp{H.PLost{m}, h, o}:
114      True{}
115    case _:
116      False{}
117
118def Drv.digest.end.go(st: W.WState, outs: List<&2, W.Out>, +wait: Bool, r: H.Resp,
119  carry: List<&2, U32>, +end: Bool) -> Dig:
120  match end:
121    case True{}:
122      Dig{W.Watcher.merge(W.Step{st, outs, wait}, W.IWatchEnd{}), r, carry, False{}}
123    case False{}:
124      Dig{W.Step{st, outs, wait}, r, carry, Bool.not(wait)}
125
126# A stream that ended cleanly is a watch timeout: the watcher resumes.
127def Drv.digest.end(s: W.Step, r: H.Resp, carry: List<&2, U32>, +end: Bool) -> Dig:
128  match s:
129    case W.Step{st, outs, +wait}:
130      Drv.digest.end.go(st, outs, wait, r, carry, end)
131
132def Drv.digest.fin(st: W.WState, r: H.Resp, cl: List<&2, U32> & List<&2, List<&2, U32>>,
133  +end: Bool) -> Dig:
134  match cl:
135    case (carry, lines):
136      Drv.digest.end(W.Watcher.steps(Drv.inputs.par(3n, lines), W.Step{st, Nil{}, False{}}), r,
137        carry, end)
138
139def Drv.digest.lines(st: W.WState, rb: H.Resp & List<&2, U32>, carry: List<&2, U32>, +eof: Bool) ->
140  Dig:
141  match rb:
142    case (+r, body):
143      Drv.digest.fin(st, r, H.Lines.split(carry, body), Bool.or(eof, Drv.ended(r)))
144
145def Drv.digest.ok(st: W.WState, r: H.Resp, carry: List<&2, U32>, +eof: Bool, +code: U32,
146  ok: Bool) -> Dig:
147  match ok:
148    case True{}:
149      Drv.digest.lines(st, H.Http.take(r), carry, eof)
150    case False{}:
151      Dig{W.Watcher.step(st, W.IWatchErr{code}), H.Http.resp.init(), Nil{}, False{}}
152
153def Drv.digest.go(st: W.WState, r: H.Resp, carry: List<&2, U32>, +eof: Bool, +code: U32) -> Dig:
154  Drv.digest.ok(st, r, carry, eof, code, U32.is_eq(code, 200))
155
156def Drv.digest(st: W.WState, +r: H.Resp, carry: List<&2, U32>, +eof: Bool) -> Dig:
157  Drv.digest.go(st, r, carry, eof, Drv.status(r))
158
159def Drv.send(chan: Chan(Msg), +src: U32, outs: List<&2, W.Out>) -> IO(Unit):
160  match outs:
161    case Nil{}:
162      IO.pure(Unit, Unit{})
163    case xs:
164      IO.bind(Result<&1, &1, Msg, Unit>, Unit, Chan.send(Msg, chan, MOuts{src, xs}),
165        r => IO.pure(Unit, Unit{}))
166
167# Hands the outputs to every subscriber (a shared watch fans out).
168def Drv.send_all(subs: List<&2, Sub>, +outs: List<&2, W.Out>) -> IO(Unit):
169  match subs:
170    case Nil{}:
171      IO.pure(Unit, Unit{})
172    case Con{Sub{ch, src}, t}:
173      IO.bind(Unit, Unit, Drv.send(ch, src, outs), u => Drv.send_all(t, outs))
174
175def Drv.jitter(+ms: U32, r: Result<&1, &1, U32 & String, U32>) -> U32:
176  match r:
177    case Done{x}:
178      (ms + (x % ((ms >> 1n) + 1)) : U32)
179    case Fail{e}:
180      ms
181
182def Drv.pause.go(+ms: U32, +fails: U32, path: String) -> IO(Unit):
183  IO.bind(Unit, Unit, L.Log.warn("list or watch failed; backing off",
184    [L.Log.kv("path", path), L.Log.kv("failures", U32.show(fails)),
185      L.Log.kv("delay_ms", U32.show(ms))]), u => IO.sleep(ms))
186
187# Sleeps the watcher's backoff, with up to 50% jitter, when asked to.
188def Drv.pause(wait: Bool, +fails: U32, path: String) -> IO(Unit):
189  match wait:
190    case True{}:
191      IO.bind(Result<&1, &1, U32 & String, U32>, Unit, IO.random_u32(),
192        r => Drv.pause.go(Drv.jitter(W.Backoff.delay(fails), r), fails, path))
193    case False{}:
194      IO.pure(Unit, Unit{})
195
196# A list from its first page (startup, or a relist after 410 Gone).
197def Drv.note_list(first: Bool, path: String) -> IO(Unit):
198  match first:
199    case True{}:
200      L.Log.info("listing", [L.Log.kv("path", path)])
201    case False{}:
202      IO.pure(Unit, Unit{})
203
204def Drv.close(s: Maybe<&1, Socket>) -> IO(Unit):
205  match s:
206    case Some{x}:
207      N.Tls.close(x)
208    case None{}:
209      IO.pure(Unit, Unit{})
210
211def WState.fails(s: W.WState) -> U32:
212  match s:
213    case W.WState{p, f, first}:
214      f
215
216def WCtx.list_req(c: WCtx, cont: String) -> H.Req:
217  match c:
218    case WCtx{cfg, api, l, fs, subs}:
219      C.Req.list(api, C.ListParams{l, fs, 500, cont, ""})
220
221def WCtx.watch_req(c: WCtx, rv: String) -> H.Req:
222  match c:
223    case WCtx{cfg, api, l, fs, subs}:
224      C.Req.watch(api, C.WatchParams{l, fs, 290, True{}}, rv)
225
226def WCtx.path(c: WCtx) -> String:
227  match c:
228    case WCtx{cfg, api, l, fs, subs}:
229      C.Api.path(api)
230
231def WCtx.cfg(c: WCtx) -> K.Config:
232  match c:
233    case WCtx{cfg, api, l, fs, subs}:
234      cfg
235
236def WCtx.subs(c: WCtx) -> List<&2, Sub>:
237  match c:
238    case WCtx{cfg, api, l, fs, subs}:
239      subs
240
241# The watcher's loop: list, then watch and stream, reconnecting and
242# relisting as the state machine says, until the fuel or the channel ends.
243def Drv.run(fuel: Nat, ev: DEv, +st: W.WState, +resp: H.Resp, +carry: List<&2, U32>,
244  +ctx: WCtx) -> IO(Unit):
245  match fuel ev:
246    case 0n e:
247      IO.pure(Unit, Unit{})
248    case 1n+f DStart{}:
249      match st:
250        case W.WState{W.WList{+c}, +fl, +first}:
251          IO.bind(Unit, Unit, Drv.note_list(first, WCtx.path(ctx)),
252            u => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
253              Cl.Client.json(WCtx.cfg(ctx), WCtx.list_req(ctx, c)),
254              r => Drv.run(f, DListed{r}, W.WState{W.WList{c}, fl, first}, resp, carry, ctx)))
255        case W.WState{W.WWatch{+rv}, +fl, +first}:
256          IO.bind(Result<&1, &1, U32 & String, N.Tls>, Unit,
257            Cl.Conn.open(WCtx.cfg(ctx), WCtx.watch_req(ctx, rv), False{}),
258            r => Drv.run(f, DOpened{r}, W.WState{W.WWatch{rv}, fl, first}, resp, carry, ctx))
259    case 1n+f DListed{Done{+j}}:
260      Drv.run(f, DStepped{W.Watcher.step(st, W.IPage{C.Page.of(j)}), None{}, False{}}, st, resp,
261        carry, ctx)
262    case 1n+f DListed{Fail{+e}}:
263      Drv.run(f, DStepped{W.Watcher.step(st, W.IListErr{Cl.KErr.code(e)}), None{}, False{}}, st,
264        resp, carry, ctx)
265    case 1n+f DOpened{Done{N.Tls{s}}}:
266      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
267        N.Tls.recv(s, 65536, 350000),
268        p => Drv.run(f, DEv.recv(p), st, H.Http.resp.init(), Nil{}, ctx))
269    case 1n+f DOpened{Fail{e}}:
270      Drv.run(f, DStepped{W.Watcher.step(st, W.IWatchErr{0}), None{}, False{}}, st, resp, carry,
271        ctx)
272    case 1n+f DRecv{s, Done{+bs}}:
273      Drv.run(f, DDig{s, Drv.digest(st, H.Http.feed(bs, resp), carry,
274        List.is_empty(&2, U32, bs))}, st, resp, carry, ctx)
275    case 1n+f DDig{s, Dig{step, r2, c2, more}}:
276      Drv.run(f, DStepped{step, Some{s}, more}, st, r2, c2, ctx)
277    case 1n+f DRecv{s, Fail{e}}:
278      Drv.run(f, DStepped{W.Watcher.step(st, W.IWatchErr{0}), Some{s}, False{}}, st, resp,
279        carry, ctx)
280    case 1n+f DStepped{W.Step{st2, outs, wait}, s, more}:
281      IO.bind(Unit, Unit, Drv.send_all(WCtx.subs(ctx), outs),
282        u => Drv.run(f, DNext{s, wait, more}, st2, resp, carry, ctx))
283    case 1n+f DNext{Some{s}, False{}, True{}}:
284      IO.bind(Socket & Result<&1, &1, U32 & String, List<&2, U32>>, Unit,
285        N.Tls.recv(s, 65536, 350000), p => Drv.run(f, DEv.recv(p), st, resp, carry, ctx))
286    case 1n+f DNext{s, wait, more}:
287      IO.bind(Unit, Unit, Drv.close(s),
288        u => IO.bind(Unit, Unit, Drv.pause(wait, WState.fails(st), WCtx.path(ctx)),
289          u2 => Drv.run(f, DStart{}, st, H.Http.resp.init(), Nil{}, ctx)))
290
291# Controller
292# ----------
293
294type CState is Data:
295  CState{store: S.Store, queue: Q.Queue, stats: Stats}
296
297type CCtx is Data:
298  CCtx{cfg: K.Config, kind: String, chan: Chan(Msg), status: Chan(SMsg), idx: U32}
299
300# A reconcile to start: the key and the object the store holds for it.
301type Job is Data:
302  Job{key: String, obj: J.Json}
303
304# What the controller loop just saw.
305type CEv is Type:
306  CMsg{m: Maybe<&1, Msg>}
307  CHandled{r: CState & List<&2, Q.Timer>}
308  CDisp{r: CState & List<&2, Job>}
309
310# The keys a primary watcher's outputs trigger: every object it applies.
311# Deletions trigger nothing; finalizers are how a reconciler sees them.
312def Ctl.self_keys(outs: List<&2, W.Out>) -> List<&2, String>:
313  match outs:
314    case Nil{}:
315      Nil{}
316    case Con{W.OApply{+o}, t}:
317      C.Obj.key(o) <> Ctl.self_keys(t)
318    case Con{W.OInitApply{+o}, t}:
319      C.Obj.key(o) <> Ctl.self_keys(t)
320    case Con{o, t}:
321      Ctl.self_keys(t)
322
323# The keys of objects a primary watcher saw deleted.
324def Ctl.gone_keys(outs: List<&2, W.Out>) -> List<&2, String>:
325  match outs:
326    case Nil{}:
327      Nil{}
328    case Con{W.ODelete{+o}, t}:
329      C.Obj.key(o) <> Ctl.gone_keys(t)
330    case Con{o, t}:
331      Ctl.gone_keys(t)
332
333def Ctl.ref_keys(rs: List<&2, C.ObjRef>) -> List<&2, String>:
334  match rs:
335    case Nil{}:
336      Nil{}
337    case Con{r, t}:
338      C.ObjRef.key(r) <> Ctl.ref_keys(t)
339
340def Ctl.out_obj(o: W.Out) -> Maybe<&2, J.Json>:
341  match o:
342    case W.OApply{x}:
343      Some{x}
344    case W.OInitApply{x}:
345      Some{x}
346    case W.ODelete{x}:
347      Some{x}
348    case _:
349      None{}
350
351def Ctl.owner_keys.put(m: Maybe<&2, J.Json>, +kind: String, rest: List<&2, String>) ->
352  List<&2, String>:
353  match m:
354    case Some{+x}:
355      List.append(&2, String, Ctl.ref_keys(C.Obj.owners(x, kind)), rest)
356    case None{}:
357      rest
358
359# The keys an owned watcher's outputs trigger: the owners of the kind
360# every changed object names (kube-runtime's Controller::owns), deletions
361# included.
362def Ctl.owner_keys(outs: List<&2, W.Out>, +kind: String) -> List<&2, String>:
363  match outs:
364    case Nil{}:
365      Nil{}
366    case Con{o, t}:
367      Ctl.owner_keys.put(Ctl.out_obj(o), kind, Ctl.owner_keys(t, kind))
368
369def Stats.count(+n: Nat) -> U32:
370  U32.from_nat(n)
371
372def Stats.events(x: Stats, +n: U32) -> Stats:
373  match x:
374    case Stats{r, e, +v}:
375      Stats{r, e, (v + n : U32)}
376
377def Stats.done(x: Stats, a: Q.Action) -> Stats:
378  match x a:
379    case Stats{+r, +e, v} Q.Retry{m}:
380      Stats{(r + 1 : U32), (e + 1 : U32), v}
381    case Stats{+r, e, v} _:
382      Stats{(r + 1 : U32), e, v}
383
384def Ctl.on_outs(st: CState, +src: U32, +outs: List<&2, W.Out>, +kind: String, own: Bool) -> CState:
385  match st own:
386    case CState{s, q, x} True{}:
387      CState{S.Store.apply_all(outs, s),
388        Q.Queue.forget_all(Ctl.gone_keys(outs), Q.Queue.add_all(Ctl.self_keys(outs), q)),
389        Stats.events(x, U32.from_nat(List.length(&2, W.Out, outs)))}
390    case CState{s, q, x} False{}:
391      CState{s, Q.Queue.add_all(Ctl.owner_keys(outs, kind), q),
392        Stats.events(x, U32.from_nat(List.length(&2, W.Out, outs)))}
393
394def Ctl.on_done(s: S.Store, x: Stats, r: Q.Queue & List<&2, Q.Timer>) ->
395  CState & List<&2, Q.Timer>:
396  match r:
397    case (q, ts):
398      (CState{s, q, x}, ts)
399
400def Ctl.on_done.st(st: CState, key: String, +a: Q.Action) -> CState & List<&2, Q.Timer>:
401  match st:
402    case CState{s, q, x}:
403      Ctl.on_done(s, Stats.done(x, a), Q.Queue.done(q, key, a))
404
405def Ctl.on_wake(st: CState, key: String, g: U32) -> CState & List<&2, Q.Timer>:
406  match st:
407    case CState{s, q, x}:
408      (CState{s, Q.Queue.wake(q, key, g), x}, Nil{})
409
410# A message's effect on the state, and the timers it arms.
411def Ctl.on_msg(st: CState, m: Msg, +kind: String) -> CState & List<&2, Q.Timer>:
412  match m:
413    case MOuts{+src, outs}:
414      (Ctl.on_outs(st, src, outs, kind, U32.is_eq(src, 0)), Nil{})
415    case MDone{key, a}:
416      Ctl.on_done.st(st, key, a)
417    case MWake{key, g}:
418      Ctl.on_wake(st, key, g)
419
420def Ctl.dispatch.drop(r: Q.Queue & List<&2, Q.Timer>, jobs: List<&2, Job>) ->
421  Q.Queue & List<&2, Job>:
422  match r:
423    case (q, ts):
424      (q, jobs)
425
426def Ctl.dispatch.put(+key: String, m: Maybe<&2, J.Json>, q: Q.Queue, jobs: List<&2, Job>) ->
427  Q.Queue & List<&2, Job>:
428  match m:
429    case Some{o}:
430      (q, Job{key, o} <> jobs)
431    case None{}:
432      # gone from the store: nothing to reconcile
433      Ctl.dispatch.drop(Q.Queue.done(q, key, Q.AwaitChange{}), jobs)
434
435def Ctl.dispatch.acc(+k: String, m: Maybe<&2, J.Json>, acc: Q.Queue & List<&2, Job>) ->
436  Q.Queue & List<&2, Job>:
437  match acc:
438    case (q, jobs):
439      Ctl.dispatch.put(k, m, q, jobs)
440
441def Ctl.dispatch.go(ks: List<&2, String>, +s: S.Store, acc: Q.Queue & List<&2, Job>) ->
442  Q.Queue & List<&2, Job>:
443  match ks:
444    case Nil{}:
445      acc
446    case Con{+k, t}:
447      Ctl.dispatch.go(t, s, Ctl.dispatch.acc(k, S.Store.get(s, k), acc))
448
449def Ctl.dispatch.wrap(s: S.Store, x: Stats, r: Q.Queue & List<&2, Job>) -> CState & List<&2, Job>:
450  match r:
451    case (q, jobs):
452      (CState{s, q, x}, jobs)
453
454def Ctl.dispatch.fin(+s: S.Store, x: Stats, r: Q.Queue & List<&2, String>) ->
455  CState & List<&2, Job>:
456  match r:
457    case (q, ks):
458      Ctl.dispatch.wrap(s, x, Ctl.dispatch.go(ks, s, (q, Nil{})))
459
460def Ctl.dispatch.ready(+s: S.Store, q: Q.Queue, x: Stats, ready: Bool) -> CState & List<&2, Job>:
461  match ready:
462    case True{}:
463      Ctl.dispatch.fin(s, x, Q.Queue.pop_all(64n, q, Nil{}))
464    case False{}:
465      (CState{s, q, x}, Nil{})
466
467# The reconciles to start now: none until the store has its first list.
468def Ctl.dispatch(st: CState) -> CState & List<&2, Job>:
469  match st:
470    case CState{+s, q, x}:
471      Ctl.dispatch.ready(s, q, x, S.Store.ready(s))
472
473def Ctl.snap(st: CState, name: String) -> Snap:
474  match st:
475    case CState{+s, +q, Stats{r, e, v}}:
476      Snap{name, S.Store.ready(s), r, e, v, Q.Queue.nrun(q), U32.from_nat(S.Store.size(s))}
477
478def Ctl.timers(ts: List<&2, Q.Timer>, +chan: Chan(Msg)) -> IO(Unit):
479  match ts:
480    case Nil{}:
481      IO.pure(Unit, Unit{})
482    case Con{Q.Timer{key, g, ms}, t}:
483      IO.bind(Unit, Unit,
484        IO.spawn(Result<&1, &1, Msg, Unit>,
485          IO.bind(Unit, Result<&1, &1, Msg, Unit>, IO.sleep(ms),
486            u => Chan.send(Msg, chan, MWake{key, g}))),
487        u => Ctl.timers(t, chan))
488
489def CCtx.chan(c: CCtx) -> Chan(Msg):
490  match c:
491    case CCtx{cfg, k, ch, st, ix}:
492      ch
493
494def CCtx.cfg(c: CCtx) -> K.Config:
495  match c:
496    case CCtx{cfg, k, ch, st, ix}:
497      cfg
498
499def CCtx.status(c: CCtx) -> Chan(SMsg):
500  match c:
501    case CCtx{cfg, k, ch, st, ix}:
502      st
503
504def CCtx.idx(c: CCtx) -> U32:
505  match c:
506    case CCtx{cfg, k, ch, st, ix}:
507      ix
508
509def CCtx.kind(c: CCtx) -> String:
510  match c:
511    case CCtx{cfg, k, ch, st, ix}:
512      k
513
514# Starts a reconcile per job; each answers MDone on the channel.
515def Ctl.spawn(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), jobs: List<&2, Job>, +chan: Chan(Msg),
516  +cfg: K.Config, +idx: U32) -> IO(Unit):
517  match jobs:
518    case Nil{}:
519      IO.pure(Unit, Unit{})
520    case Con{Job{key, o}, t}:
521      IO.bind(Unit, Unit,
522        IO.spawn(Result<&1, &1, Msg, Unit>,
523          IO.bind(Q.Action, Result<&1, &1, Msg, Unit>, rec(idx, o, cfg),
524            a => Chan.send(Msg, chan, MDone{key, a}))),
525        u => Ctl.spawn(~rec, t, chan, cfg, idx))
526
527# The controller loop: a message in, the state stepped, timers armed,
528# reconciles started, the next message awaited.
529def Ctl.run(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), fuel: Nat, ev: CEv, +st: CState,
530  +ctx: CCtx) -> IO(Unit):
531  match fuel ev:
532    case 0n e:
533      IO.pure(Unit, Unit{})
534    case 1n+f CMsg{None{}}:
535      IO.pure(Unit, Unit{})
536    case 1n+f CMsg{Some{m}}:
537      Ctl.run(~rec, f, CHandled{Ctl.on_msg(st, m, CCtx.kind(ctx))}, st, ctx)
538    case 1n+f CHandled{(+st1, ts)}:
539      IO.bind(Unit, Unit, Ctl.timers(ts, CCtx.chan(ctx)),
540        u => Ctl.run(~rec, f, CDisp{Ctl.dispatch(st1)}, st1, ctx))
541    case 1n+f CDisp{(+st2, jobs)}:
542      IO.bind(Unit, Unit, Ctl.spawn(~rec, jobs, CCtx.chan(ctx), CCtx.cfg(ctx), CCtx.idx(ctx)),
543        u => IO.bind(Result<&1, &1, SMsg, Unit>, Unit,
544          Chan.send(SMsg, CCtx.status(ctx), SPut{CCtx.idx(ctx), Ctl.snap(st2, U32.show(CCtx.idx(ctx)) ++ "/" ++ CCtx.kind(ctx))}),
545          u2 => IO.bind(Maybe<&1, Msg>, Unit, Chan.recv(Msg, CCtx.chan(ctx)),
546            m => Ctl.run(~rec, f, CMsg{m}, st2, ctx))))
547
548def Ctl.watch.subs(+cfg: K.Config, api: C.Api, label: String, subs: List<&2, Sub>) -> IO(Unit):
549  IO.spawn(Unit, Drv.run(4294967295n, DStart{}, W.WState.init(), H.Http.resp.init(), Nil{},
550    WCtx{cfg, api, label, "", subs}))
551
552def Ctl.watch(+cfg: K.Config, api: C.Api, label: String, +src: U32, +chan: Chan(Msg)) -> IO(Unit):
553  IO.spawn(Unit, Drv.run(4294967295n, DStart{}, W.WState.init(), H.Http.resp.init(), Nil{},
554    WCtx{cfg, api, label, "", [Sub{chan, src}]}))
555
556def Ctl.watch_all(apis: List<&2, C.Api>, +n: U32, +cfg: K.Config, +chan: Chan(Msg)) -> IO(Unit):
557  match apis:
558    case Nil{}:
559      IO.pure(Unit, Unit{})
560    case Con{a, t}:
561      IO.bind(Unit, Unit, Ctl.watch(cfg, a, "", n, chan),
562        u => Ctl.watch_all(t, (n + 1 : U32), cfg, chan))
563
564def Api.kind(a: C.Api) -> String:
565  match a:
566    case C.Api{r, s}:
567      C.Res.kind(r)
568
569def Keeper.all.put(r: Map<&2, Snap> & Snap, rest: List<&2, Snap>) -> List<&2, Snap>:
570  match r:
571    case (m, s):
572      s <> rest
573
574def Keeper.all.go(ks: List<&2, String>, +m: Map<&2, Snap>) -> List<&2, Snap>:
575  match ks:
576    case Nil{}:
577      Nil{}
578    case Con{k, t}:
579      Keeper.all.put(Map.get(Snap, Snap{"", False{}, 0, 0, 0, 0, 0}, m, k), Keeper.all.go(t, m))
580
581def Keeper.all(+m: Map<&2, Snap>) -> List<&2, Snap>:
582  Keeper.all.go(Map.keys(&2, Snap, m), m)
583
584# The status keeper: holds each controller's latest snapshot and answers
585# queries at once.
586def Keeper.run(fuel: Nat, m: Maybe<&1, SMsg>, +last: Map<&2, Snap>, +inbox: Chan(SMsg)) -> IO(Unit):
587  match fuel m:
588    case 0n x:
589      IO.pure(Unit, Unit{})
590    case 1n+f None{}:
591      IO.pure(Unit, Unit{})
592    case 1n+f Some{SPut{i, s}}:
593      IO.bind(Maybe<&1, SMsg>, Unit, Chan.recv(SMsg, inbox),
594        x => Keeper.run(f, x, Map.set(&2, Snap, last, U32.show(i), s), inbox))
595    case 1n+f Some{SGet{r}}:
596      IO.bind(Result<&1, &1, List<&2, Snap>, Unit>, Unit,
597        Chan.send(List<&2, Snap>, r, Keeper.all(last)),
598        u => IO.bind(Maybe<&1, SMsg>, Unit, Chan.recv(SMsg, inbox), x => Keeper.run(f, x, last, inbox)))
599
600# Starts a status keeper; answers its inbox.
601def Keeper.start() -> IO(Chan(SMsg)):
602  IO.bind(Chan(SMsg), Chan(SMsg), Chan.new(SMsg, 4096),
603    +c => IO.bind(Unit, Chan(SMsg), IO.spawn(Unit,
604      IO.bind(Maybe<&1, SMsg>, Unit, Chan.recv(SMsg, c),
605        x => Keeper.run(4294967295n, x, Map.new(&2, Snap), c))), u => IO.pure(Chan(SMsg), c)))
606
607def Ctl.loop(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +primary: C.Api, +max: U32,
608  +chan: Chan(Msg), +status: Chan(SMsg), +idx: U32) -> IO(Unit):
609  Ctl.run(~rec, 4294967295n,
610    CDisp{(CState{S.Store.new(), Q.Queue.new(max), Stats{0, 0, 0}}, Nil{})},
611    CState{S.Store.new(), Q.Queue.new(max), Stats{0, 0, 0}},
612    CCtx{cfg, Api.kind(primary), chan, status, idx})
613
614# Runs a controller (kube-runtime's Controller::new(..).owns(..).run(..)):
615# primary is watched (only objects matching the label selector, "" for
616# all) and reconciled, each of owned is watched and mapped to
617# its owners of primary's kind; at most max reconciles run at once. rec
618# answers what to do next for the object, as kube-runtime's Action. chan
619# is the controller's inbox (Controller.inbox); status receives a snapshot
620# after every step (Keeper.start).
621def Controller.start(~rec: J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +primary: C.Api,
622  label: String, owned: List<&2, C.Api>, +max: U32, +chan: Chan(Msg), +status: Chan(SMsg)) ->
623  IO(Unit):
624  do IO<Unit>:
625    Ctl.watch(cfg, primary, label, 0, chan)
626    Ctl.watch_all(owned, 1, cfg, chan)
627    Ctl.loop(~(i => o => c => rec(o, c)), cfg, primary, max, chan, status, 0)
628
629def Controller.inbox() -> IO(Chan(Msg)):
630  Chan.new(Msg, 4096)
631
632# Manager
633# -------
634#
635# Several controllers in one process (controller-runtime's Manager, with
636# kube-runtime's shared reflectors): each runs its own loop, store and
637# queue, but a watch they share (the same resource, scope and label
638# selector) runs once, and its outputs fan out to every controller that
639# subscribed, so the API server sees one list and one watch per resource.
640
641# A controller of a manager: its primary watch (with a label selector), the
642# kinds it owns, and how many reconciles it runs at once. A manager's
643# reconcile gets the controller's index (its place in the list) first.
644type Ctrl is Data:
645  Ctrl{primary: C.Api, label: String, owned: List<&2, C.Api>, max: U32}
646
647type Watch is Data:
648  Watch{api: C.Api, label: String, subs: List<&2, Sub>}
649
650def Watch.key(+api: C.Api, +label: String) -> String:
651  C.Api.path(api) ++ "?" ++ label
652
653def Plan.hit(ws: List<&2, Watch>, +key: String) -> Bool:
654  match ws:
655    case Con{Watch{+a, +l, s}, t}:
656      String.eq(Watch.key(a, l), key)
657    case _:
658      False{}
659
660# Adds a subscription: to the watch of the same key, or as a new watch.
661def Plan.add(ws: List<&2, Watch>, +api: C.Api, +label: String, +sub: Sub, hit: Bool) -> List<&2, Watch>:
662  match ws hit:
663    case Con{Watch{a, l, ss}, t} True{}:
664      Watch{a, l, List.append(&2, Sub, ss, [sub])} <> t
665    case Con{+w, +t} False{}:
666      w <> Plan.add(t, api, label, sub, Plan.hit(t, Watch.key(api, label)))
667    case Nil{} _:
668      [Watch{api, label, [sub]}]
669
670def Plan.sub(+ws: List<&2, Watch>, +api: C.Api, +label: String, +sub: Sub) -> List<&2, Watch>:
671  Plan.add(ws, api, label, sub, Plan.hit(ws, Watch.key(api, label)))
672
673def Plan.owned(os: List<&2, C.Api>, +ch: Chan(Msg), +n: U32, ws: List<&2, Watch>) -> List<&2, Watch>:
674  match os:
675    case Nil{}:
676      ws
677    case Con{a, t}:
678      Plan.owned(t, ch, (n + 1 : U32), Plan.sub(ws, a, "", Sub{ch, n}))
679
680# Every watch the controllers need, each once, with all its subscribers.
681def Plan.build(cs: List<&2, Ctrl>, chs: List<&2, Chan(Msg)>, ws: List<&2, Watch>) -> List<&2, Watch>:
682  match cs chs:
683    case Con{Ctrl{p, l, os, m}, t} Con{+ch, ct}:
684      Plan.build(t, ct, Plan.owned(os, ch, 1, Plan.sub(ws, p, l, Sub{ch, 0})))
685    case _ _:
686      ws
687
688def Manager.inboxes(cs: List<&2, Ctrl>) -> IO(List<&2, Chan(Msg)>):
689  match cs:
690    case Nil{}:
691      IO.pure(List<&2, Chan(Msg)>, Nil{})
692    case Con{c, t}:
693      IO.bind(Chan(Msg), List<&2, Chan(Msg)>, Controller.inbox(),
694        ch => IO.bind(List<&2, Chan(Msg)>, List<&2, Chan(Msg)>, Manager.inboxes(t),
695          rest => IO.pure(List<&2, Chan(Msg)>, ch <> rest)))
696
697def Manager.watch(+cfg: K.Config, ws: List<&2, Watch>) -> IO(Unit):
698  match ws:
699    case Nil{}:
700      IO.pure(Unit, Unit{})
701    case Con{Watch{a, l, ss}, t}:
702      IO.bind(Unit, Unit, Ctl.watch.subs(cfg, a, l, ss), u => Manager.watch(cfg, t))
703
704def Manager.loops(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, cs: List<&2, Ctrl>,
705  chs: List<&2, Chan(Msg)>, +status: Chan(SMsg), +i: U32) -> IO(Unit):
706  match cs chs:
707    case Con{Ctrl{p, l, os, m}, t} Con{ch, ct}:
708      IO.bind(Unit, Unit, IO.spawn(Unit, Ctl.loop(~rec, cfg, p, m, ch, status, i)),
709        u => Manager.loops(~rec, cfg, t, ct, status, (i + 1 : U32)))
710    case _ _:
711      IO.pure(Unit, Unit{})
712
713# Runs every controller, sharing their watches. rec(i, obj, cfg) reconciles
714# obj for controller i.
715def Manager.start(~rec: U32 -> J.Json -> K.Config -> IO(Q.Action), +cfg: K.Config, +cs: List<&2, Ctrl>,
716  +status: Chan(SMsg)) -> IO(Unit):
717  IO.bind(List<&2, Chan(Msg)>, Unit, Manager.inboxes(cs),
718    +chs => IO.bind(Unit, Unit, Manager.watch(cfg, Plan.build(cs, chs, Nil{})),
719      u => Manager.loops(~rec, cfg, cs, chs, status, 0)))