1	import Base
2	import ./json.bend as J
3	import ./core.bend as C
4	import ./config.bend as K
5	import ./client.bend as Cl
6	import ./time.bend as T
7	import ./sys.bend as Y
8	import ./log.bend as L
9	import ./lease.bend as E
10	
11	# Leader
12	# ======
13	#
14	# client-go's leaderelection loop over lease.bend's decisions, with its
15	# defaults: a 15s lease, renewed every 2s, given up when no renewal lands
16	# for 10s. Losing the lease exits the process (status 1), as client-go's
17	# usual OnStoppedLeading does, so two leaders never act at once; SIGTERM
18	# hands the lease back (status 0).
19	
20	type LOpts is Data:
21	  LOpts{ns: String, name: String, id: String, dur: U32, deadline: U32, retry: U32}
22	
23	def LOpts.default(ns: String, name: String, id: String) -> LOpts:
24	  LOpts{ns, name, id, 15, 10000, 2000}
25	
26	def LOpts.api(o: LOpts) -> C.Api:
27	  match o:
28	    case LOpts{ns, n, i, d, dl, r}:
29	      C.Api{C.Res.leases(), C.InNs{ns}}
30	
31	def LOpts.name(o: LOpts) -> String:
32	  match o:
33	    case LOpts{ns, n, i, d, dl, r}:
34	      n
35	
36	def LOpts.retry(o: LOpts) -> U32:
37	  match o:
38	    case LOpts{ns, n, i, d, dl, r}:
39	      r
40	
41	def LOpts.deadline(o: LOpts) -> U32:
42	  match o:
43	    case LOpts{ns, n, i, d, dl, r}:
44	      dl
45	
46	def Leader.read.err(e: Cl.KErr, missing: Bool) -> Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>:
47	  match missing:
48	    case True{}:
49	      Done{None{}}
50	    case False{}:
51	      Fail{e}
52	
53	# The Lease read: Some, None on 404, or the error.
54	def Leader.read(r: Result<&2, &2, Cl.KErr, J.Json>) -> Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>:
55	  match r:
56	    case Done{l}:
57	      Done{Some{l}}
58	    case Fail{+e}:
59	      Leader.read.err(e, U32.is_eq(Cl.KErr.code(e), 404))
60	
61	def Leader.decide(r: Result<&2, &2, Cl.KErr, Maybe<&2, J.Json>>, +o: LOpts, +obs: E.Obs,
62	  +w: T.Wall) -> Result<&2, &2, Cl.KErr, E.Decision>:
63	  match r o:
64	    case Done{cur} LOpts{ns, n, id, d, dl, rt}:
65	      Done{E.Lease.decide(cur, ns, n, id, obs, T.Wall.ms(w), d, T.Wall.micro(w))}
66	    case Fail{e} _:
67	      Fail{e}
68	
69	# What the loop just saw.
70	type LEv is Type:
71	  LTry{}
72	  LRead{w: T.Wall, r: Result<&2, &2, Cl.KErr, J.Json>}
73	  LDecided{now: U32, d: Result<&2, &2, Cl.KErr, E.Decision>}
74	  LWrote{now: U32, ok: Bool, late: Bool}
75	  LFail{stop: Bool}
76	
77	def Leader.wrote(r: Result<&2, &2, Cl.KErr, J.Json>) -> Bool:
78	  Result.is_done(&2, &2, Cl.KErr, J.Json, r)
79	
80	def Leader.late(+now: U32, +last: U32, +deadline: U32) -> Bool:
81	  U32.is_gt((now - last : U32), deadline)
82	
83	# On first acquiring: log and start the work.
84	def Leader.started(leading: Bool, start: Maybe<&1, IO(Unit)>) -> IO(Unit):
85	  match leading start:
86	    case False{} Some{act}:
87	      IO.bind(Unit, Unit, L.Log.info("became leader", Nil{}), u => IO.spawn(Unit, act))
88	    case _ _:
89	      IO.pure(Unit, Unit{})
90	
91	# A leader that cannot renew within the deadline stops at once (a
92	# candidate that fails just tries again).
93	def Leader.lost() -> IO(Unit):
94	  IO.bind(Unit, Unit, L.Log.error("leader lease lost; exiting", Nil{}), u => IO.exit(1))
95	
96	# The election loop. start runs once, when the lease is first ours.
97	def Leader.run(fuel: Nat, ev: LEv, +cfg: K.Config, +o: LOpts, +obs: E.Obs, +leading: Bool,
98	  +last: U32, start: Maybe<&1, IO(Unit)>) -> IO(Unit):
99	  match fuel ev:
100	    case 0n e:
101	      IO.pure(Unit, Unit{})
102	    case 1n+f LTry{}:
103	      IO.bind(U32 & (U32 & U32), Unit, Y.Sys.wall(),
104	        w => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
105	          Cl.Client.get(cfg, LOpts.api(o), LOpts.name(o)),
106	          r => Leader.run(f, LRead{T.Wall.of(w), r}, cfg, o, obs, leading, last, start)))
107	    case 1n+f LRead{+w, r}:
108	      Leader.run(f, LDecided{T.Wall.ms(w), Leader.decide(Leader.read(r), o, obs, w)}, cfg, o, obs,
109	        leading, last, start)
110	    case 1n+f LDecided{+now, Done{E.Decision{E.LCreate{l}, ob}}}:
111	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit, Cl.Client.create(cfg, LOpts.api(o), l),
112	        r => Leader.run(f, LWrote{now, Leader.wrote(r), Leader.late(now, last, LOpts.deadline(o))},
113	          cfg, o, ob, leading, last, start))
114	    case 1n+f LDecided{+now, Done{E.Decision{E.LUpdate{l}, ob}}}:
115	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
116	        Cl.Client.replace(cfg, LOpts.api(o), LOpts.name(o), l),
117	        r => Leader.run(f, LWrote{now, Leader.wrote(r), Leader.late(now, last, LOpts.deadline(o))},
118	          cfg, o, ob, leading, last, start))
119	    case 1n+f LDecided{+now, Done{E.Decision{E.LWait{}, ob}}}:
120	      Leader.run(f, LWrote{now, False{}, Leader.late(now, last, LOpts.deadline(o))}, cfg, o, ob,
121	        leading, last, start)
122	    case 1n+f LDecided{+now, Fail{e}}:
123	      IO.bind(Unit, Unit, L.Log.warn("lease read failed", [L.Log.kv("error", Cl.KErr.show(e))]),
124	        u => Leader.run(f, LWrote{now, False{}, Leader.late(now, last, LOpts.deadline(o))}, cfg, o,
125	          obs, leading, last, start))
126	    case 1n+f LWrote{now, True{}, late}:
127	      IO.bind(Unit, Unit, Leader.started(leading, start),
128	        u => IO.bind(Unit, Unit, IO.sleep(LOpts.retry(o)),
129	          u2 => Leader.run(f, LTry{}, cfg, o, obs, True{}, now, None{})))
130	    case 1n+f LWrote{now, False{}, late}:
131	      Leader.run(f, LFail{Bool.and(leading, late)}, cfg, o, obs, leading, last, start)
132	    case 1n+f LFail{True{}}:
133	      Leader.lost()
134	    case 1n+f LFail{False{}}:
135	      IO.bind(Unit, Unit, IO.sleep(LOpts.retry(o)),
136	        u => Leader.run(f, LTry{}, cfg, o, obs, leading, last, start))
137	
138	def Leader.release.mine(+cfg: K.Config, +o: LOpts, l: J.Json, mine: Bool) -> IO(Unit):
139	  match mine:
140	    case True{}:
141	      IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
142	        Cl.Client.replace(cfg, LOpts.api(o), LOpts.name(o), l),
143	        r => L.Log.info("released leader lease", Nil{}))
144	    case False{}:
145	      IO.pure(Unit, Unit{})
146	
147	# Hands the lease back if it is ours.
148	def Leader.release.got(r: Result<&2, &2, Cl.KErr, J.Json>, +cfg: K.Config, +o: LOpts,
149	  w: T.Wall) -> IO(Unit):
150	  match r o:
151	    case Done{+l} LOpts{ns, n, +id, d, dl, rt}:
152	      Leader.release.mine(cfg, LOpts{ns, n, id, d, dl, rt}, E.Lease.release(l, T.Wall.micro(w)),
153	        String.eq(E.Lease.holder(l), id))
154	    case _ _:
155	      IO.pure(Unit, Unit{})
156	
157	def Leader.release(+cfg: K.Config, +o: LOpts) -> IO(Unit):
158	  IO.bind(U32 & (U32 & U32), Unit, Y.Sys.wall(),
159	    w => IO.bind(Result<&2, &2, Cl.KErr, J.Json>, Unit,
160	      Cl.Client.get(cfg, LOpts.api(o), LOpts.name(o)), r => Leader.release.got(r, cfg, o, T.Wall.of(w))))
161	
162	# On SIGTERM or SIGINT: release the lease, then exit 0.
163	def Leader.on_signal(+cfg: K.Config, +o: LOpts) -> IO(Unit):
164	  IO.bind(U32, Unit, Y.Sys.signal(),
165	    sig => IO.bind(Unit, Unit, L.Log.info("terminating", [L.Log.kv("signal", U32.show(sig))]),
166	      u => IO.bind(Unit, Unit, Leader.release(cfg, o), u2 => IO.exit(0))))
167	
168	# Runs work only while we hold the lease (client-go's RunOrDie).
169	def Leader.elect(+cfg: K.Config, +o: LOpts, work: IO(Unit)) -> IO(Unit):
170	  IO.bind(Unit, Unit, IO.spawn(Unit, Leader.on_signal(cfg, o)),
171	    u => Leader.run(4294967295n, LTry{}, cfg, o, E.Obs.none(), False{}, 0, Some{work}))