1	import Base
2	import ./json.bend as J
3	import ./utf8.bend as U
4	import ./http.bend as H
5	import ./core.bend as C
6	import ./config.bend as K
7	import ./net.bend as N
8	import ./client.bend as Cl
9	import ./watcher.bend as W
10	import ./store.bend as S
11	import ./queue.bend as Q
12	import ./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	
27	type 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.
33	type 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).
40	type SMsg is Data:
41	  SPut{idx: U32, snap: Snap}
42	  SGet{reply: Chan(List<&2, Snap>)}
43	
44	# Counters since start.
45	type 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).
53	type Sub is Data:
54	  Sub{chan: Chan(Msg), src: U32}
55	
56	type 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.
60	type 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	
69	def 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.
75	def 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.
87	def 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.
97	type Dig is Data:
98	  Dig{step: W.Step, resp: H.Resp, carry: List<&2, U32>, more: Bool}
99	
100	def 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	
107	def 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	
118	def 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.
127	def 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	
132	def 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	
139	def 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	
145	def 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	
153	def 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	
156	def 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	
159	def 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).
168	def 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	
175	def 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	
182	def 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.
188	def 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).
197	def 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	
204	def 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	
211	def WState.fails(s: W.WState) -> U32:
212	  match s:
213	    case W.WState{p, f, first}:
214	      f
215	
216	def 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	
221	def 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	
226	def WCtx.path(c: WCtx) -> String:
227	  match c:
228	    case WCtx{cfg, api, l, fs, subs}:
229	      C.Api.path(api)
230	
231	def WCtx.cfg(c: WCtx) -> K.Config:
232	  match c:
233	    case WCtx{cfg, api, l, fs, subs}:
234	      cfg
235	
236	def 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.
243	def 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	
294	type CState is Data:
295	  CState{store: S.Store, queue: Q.Queue, stats: Stats}
296	
297	type 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.
301	type Job is Data:
302	  Job{key: String, obj: J.Json}
303	
304	# What the controller loop just saw.
305	type 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.
312	def 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.
324	def 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	
333	def 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	
340	def 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	
351	def 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.
362	def 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	
369	def Stats.count(+n: Nat) -> U32:
370	  U32.from_nat(n)
371	
372	def Stats.events(x: Stats, +n: U32) -> Stats:
373	  match x:
374	    case Stats{r, e, +v}:
375	      Stats{r, e, (v + n : U32)}
376	
377	def 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	
384	def 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	
394	def 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	
400	def 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	
405	def 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.
411	def 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	
420	def 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	
426	def 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	
435	def 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	
441	def 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	
449	def 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	
454	def 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	
460	def 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.
468	def 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	
473	def 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	
478	def 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	
489	def CCtx.chan(c: CCtx) -> Chan(Msg):
490	  match c:
491	    case CCtx{cfg, k, ch, st, ix}:
492	      ch
493	
494	def CCtx.cfg(c: CCtx) -> K.Config:
495	  match c:
496	    case CCtx{cfg, k, ch, st, ix}:
497	      cfg
498	
499	def CCtx.status(c: CCtx) -> Chan(SMsg):
500	  match c:
501	    case CCtx{cfg, k, ch, st, ix}:
502	      st
503	
504	def CCtx.idx(c: CCtx) -> U32:
505	  match c:
506	    case CCtx{cfg, k, ch, st, ix}:
507	      ix
508	
509	def 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.
515	def 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.
529	def 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	
548	def 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	
552	def 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	
556	def 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	
564	def Api.kind(a: C.Api) -> String:
565	  match a:
566	    case C.Api{r, s}:
567	      C.Res.kind(r)
568	
569	def 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	
574	def 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	
581	def 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.
586	def 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.
601	def 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	
607	def 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).
621	def 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	
629	def 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.
644	type Ctrl is Data:
645	  Ctrl{primary: C.Api, label: String, owned: List<&2, C.Api>, max: U32}
646	
647	type Watch is Data:
648	  Watch{api: C.Api, label: String, subs: List<&2, Sub>}
649	
650	def Watch.key(+api: C.Api, +label: String) -> String:
651	  C.Api.path(api) ++ "?" ++ label
652	
653	def 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.
661	def 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	
670	def 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	
673	def 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.
681	def 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	
688	def 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	
697	def 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	
704	def 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.
715	def 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)))