1	import Base
2	import ./json.bend as J
3	import ./core.bend as C
4	
5	# Watcher
6	# =======
7	#
8	# kube-runtime's watcher as a pure state machine, with client-go's
9	# reflector deciding how the API behaves:
10	#
11	# - A watcher starts by listing in pages (limit + continue) and then watches
12	#   from the list's resourceVersion, with bookmarks, resuming from the last
13	#   version it saw whenever a watch ends (the server's timeoutSeconds).
14	# - A 410 Gone (an expired version, on a watch or a continue token) relists
15	#   from scratch at once, as the reflector does; the store swaps in the new
16	#   list whole, so objects deleted meanwhile disappear.
17	# - Other failures retry the same step after an exponential backoff
18	#   (800ms doubling to 30s, as both kube-rs and client-go default to);
19	#   success resets it.
20	# - resourceVersion is opaque: it is only ever copied, never compared.
21	#
22	# The IO driver (runtime.bend) performs the action a state asks for, feeds
23	# the outcome back as an input, and hands the outputs to the store.
24	
25	# What to do next: list (from a continue token, "" for the first page) or
26	# watch from a version.
27	type Phase is Data:
28	  WList{cont: String}
29	  WWatch{rv: String}
30	
31	# fails counts failures since the last success; first says the next page
32	# starts a fresh list.
33	type WState is Data:
34	  WState{phase: Phase, fails: U32, first: Bool}
35	
36	type Input is Data:
37	  IPage{page: C.Page}
38	  IListErr{code: U32}
39	  IEvent{ev: C.WEvent}
40	  IWatchEnd{}
41	  IWatchErr{code: U32}
42	
43	# What the store and controller see (kube-runtime's watcher::Event).
44	type Out is Data:
45	  OInit{}
46	  OInitApply{obj: J.Json}
47	  OInitDone{}
48	  OApply{obj: J.Json}
49	  ODelete{obj: J.Json}
50	
51	# A step's result: the next state, its outputs in order, and whether the
52	# driver must back off before the next action.
53	type Step is Data:
54	  Step{state: WState, outs: List<&2, Out>, wait: Bool}
55	
56	def WState.init() -> WState:
57	  WState{WList{""}, 0, True{}}
58	
59	def Backoff.delay.go(+fails: U32, capped: Bool) -> U32:
60	  match capped:
61	    case True{}:
62	      30000
63	    case False{}:
64	      U32.min((800 << U32.to_nat((fails - 1 : U32)) : U32), 30000)
65	
66	# Backoff
67	# -------
68	
69	# The delay before the next try after fails failures: 800ms doubling, at
70	# most 30s. The driver adds jitter.
71	def Backoff.delay(+fails: U32) -> U32:
72	  Backoff.delay.go(fails, U32.is_ge(fails, 6))
73	
74	# Steps
75	# -----
76	
77	def Watcher.page_outs(xs: List<&2, J.Json>, tail: List<&2, Out>) -> List<&2, Out>:
78	  match xs:
79	    case Nil{}:
80	      tail
81	    case Con{x, t}:
82	      OInitApply{x} <> Watcher.page_outs(t, tail)
83	
84	def Watcher.init_out(first: Bool, rest: List<&2, Out>) -> List<&2, Out>:
85	  match first:
86	    case True{}:
87	      OInit{} <> rest
88	    case False{}:
89	      rest
90	
91	def Watcher.done_out(last: Bool) -> List<&2, Out>:
92	  match last:
93	    case True{}:
94	      [OInitDone{}]
95	    case False{}:
96	      Nil{}
97	
98	def Watcher.page_phase(rv: String, +cont: String, last: Bool) -> Phase:
99	  match last:
100	    case True{}:
101	      WWatch{rv}
102	    case False{}:
103	      WList{cont}
104	
105	def Watcher.page.go(first: Bool, items: List<&2, J.Json>, rv: String, +cont: String,
106	  +last: Bool) -> Step:
107	  Step{WState{Watcher.page_phase(rv, cont, last), 0, last},
108	    Watcher.init_out(first, Watcher.page_outs(items, Watcher.done_out(last))), False{}}
109	
110	# A page of the list: the first page opens a relist (Init), the last one
111	# (no continue token) closes it (InitDone) and starts the watch.
112	def Watcher.page(first: Bool, p: C.Page) -> Step:
113	  match p:
114	    case C.Page{items, rv, +cont}:
115	      Watcher.page.go(first, items, rv, cont, String.is_empty(cont))
116	
117	def Watcher.is_gone(+code: U32) -> Bool:
118	  U32.is_eq(code, 410)
119	
120	def Watcher.relist(+fails: U32) -> Step:
121	  Step{WState{WList{""}, fails, True{}}, Nil{}, False{}}
122	
123	def Watcher.retry(p: Phase, +fails: U32, first: Bool) -> Step:
124	  Step{WState{p, (fails + 1 : U32), first}, Nil{}, True{}}
125	
126	# A failed list: an expired continue token relists; else the same page is
127	# retried after a backoff.
128	def Watcher.list_err(cont: String, +fails: U32, first: Bool, gone: Bool) -> Step:
129	  match gone:
130	    case True{}:
131	      Watcher.relist(fails)
132	    case False{}:
133	      Watcher.retry(WList{cont}, fails, first)
134	
135	def Watcher.watch_err(rv: String, +fails: U32, gone: Bool) -> Step:
136	  match gone:
137	    case True{}:
138	      Watcher.relist(fails)
139	    case False{}:
140	      Watcher.retry(WWatch{rv}, fails, False{})
141	
142	# One watch event while watching from rv.
143	def Watcher.event(rv: String, +fails: U32, ev: C.WEvent) -> Step:
144	  match ev:
145	    case C.WAdded{+o}:
146	      Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [OApply{o}], False{}}
147	    case C.WModified{+o}:
148	      Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [OApply{o}], False{}}
149	    case C.WDeleted{+o}:
150	      Step{WState{WWatch{C.Obj.rv(o)}, 0, False{}}, [ODelete{o}], False{}}
151	    case C.WBookmark{v}:
152	      Step{WState{WWatch{v}, 0, False{}}, Nil{}, False{}}
153	    case C.WError{C.Status{+code, r, m}}:
154	      Watcher.watch_err(rv, fails, Watcher.is_gone(code))
155	    case C.WBad{m}:
156	      Watcher.retry(WWatch{rv}, fails, False{})
157	
158	# The state machine.
159	def Watcher.step(st: WState, i: Input) -> Step:
160	  match st i:
161	    case WState{WList{c}, f, first} IPage{p}:
162	      Watcher.page(first, p)
163	    case WState{WList{c}, +f, first} IListErr{+code}:
164	      Watcher.list_err(c, f, first, Watcher.is_gone(code))
165	    case WState{WWatch{rv}, +f, first} IEvent{ev}:
166	      Watcher.event(rv, f, ev)
167	    case WState{WWatch{rv}, f, first} IWatchEnd{}:
168	      Step{WState{WWatch{rv}, 0, False{}}, Nil{}, False{}}
169	    case WState{WWatch{rv}, +f, first} IWatchErr{+code}:
170	      Watcher.watch_err(rv, f, Watcher.is_gone(code))
171	    case s _:
172	      # an input the phase did not ask for changes nothing
173	      Step{s, Nil{}, False{}}
174	
175	def Watcher.merge.go(outs: List<&2, Out>, w: Bool, s: Step) -> Step:
176	  match s:
177	    case Step{st, more, w2}:
178	      Step{st, List.append(&2, Out, outs, more), Bool.or(w, w2)}
179	
180	# Feeds a run of inputs (the events of one recv), joining the outputs. A
181	# step that asks to back off ends the run's effect on the stream: the
182	# driver reconnects.
183	def Watcher.merge(acc: Step, x: Input) -> Step:
184	  match acc:
185	    case Step{st, outs, w}:
186	      Watcher.merge.go(outs, w, Watcher.step(st, x))
187	
188	def Watcher.steps(xs: List<&2, Input>, acc: Step) -> Step:
189	  match xs:
190	    case Nil{}:
191	      acc
192	    case Con{x, t}:
193	      Watcher.steps(t, Watcher.merge(acc, x))