1	import Base
2	import ./json.bend as J
3	import ./core.bend as C
4	
5	# Lease
6	# =====
7	#
8	# Leader election over a coordination.k8s.io/v1 Lease, deciding as
9	# client-go's LeaderElector.tryAcquireOrRenew does:
10	#
11	# - No Lease: create one naming us.
12	# - A Lease held by someone else whose record we saw change less than
13	#   leaseDuration ago (by our own clock: remote timestamps are never
14	#   compared, so clock skew cannot steal a lease): wait.
15	# - Otherwise (ours, unheld, or expired): write it naming us, through the
16	#   resourceVersion the read returned, so of two contenders only one
17	#   write lands (optimistic concurrency).
18	#
19	# Taking a lease from another holder counts a transition and resets
20	# acquireTime; renewing our own keeps both.
21	
22	# What we last saw of the record, and when (our clock, in ms).
23	type Obs is Data:
24	  Obs{holder: String, renew: String, at: U32}
25	
26	type LeaseAct is Data:
27	  LCreate{lease: J.Json}
28	  LUpdate{lease: J.Json}
29	  LWait{}
30	
31	type Decision is Data:
32	  Decision{act: LeaseAct, obs: Obs}
33	
34	def Obs.none() -> Obs:
35	  Obs{"", "", 0}
36	
37	def Lease.spec(+l: J.Json, field: String) -> J.Json:
38	  J.Json.get(J.Json.get(l, "spec"), field)
39	
40	def Lease.holder(+l: J.Json) -> String:
41	  J.Json.str_or(Lease.spec(l, "holderIdentity"), "")
42	
43	def Lease.renew_time(+l: J.Json) -> String:
44	  J.Json.str_or(Lease.spec(l, "renewTime"), "")
45	
46	def Lease.transitions(+l: J.Json) -> U32:
47	  Maybe.default(&2, U32, J.Json.as_u32(Lease.spec(l, "leaseTransitions")), 0)
48	
49	def Lease.new_spec(id: String, +dur: U32, acquire: String, renew: String, +trans: U32) -> J.Json:
50	  J.Json.obj([J.KV{"holderIdentity", J.JStr{id}}, J.KV{"leaseDurationSeconds", J.Json.u32(dur)},
51	    J.KV{"acquireTime", J.JStr{acquire}}, J.KV{"renewTime", J.JStr{renew}},
52	    J.KV{"leaseTransitions", J.Json.u32(trans)}])
53	
54	# A fresh Lease naming us.
55	def Lease.fresh(ns: String, name: String, id: String, +dur: U32, +now: String) -> J.Json:
56	  J.Json.obj([J.KV{"apiVersion", J.JStr{"coordination.k8s.io/v1"}}, J.KV{"kind", J.JStr{"Lease"}},
57	    J.KV{"metadata", J.Json.obj([J.KV{"name", J.JStr{name}}, J.KV{"namespace", J.JStr{ns}}])},
58	    J.KV{"spec", Lease.new_spec(id, dur, now, now, 0)}])
59	
60	def Lease.observe.go(h: String, r: String, o: Obs, +now: U32, same: Bool) -> Obs:
61	  match same:
62	    case True{}:
63	      o
64	    case False{}:
65	      Obs{h, r, now}
66	
67	# The record seen now: the old observation if nothing changed, else a new
68	# one stamped now.
69	def Lease.observe(+l: J.Json, o: Obs, +now: U32) -> Obs:
70	  match o:
71	    case Obs{+h, +r, at}:
72	      Lease.observe.go(Lease.holder(l), Lease.renew_time(l), Obs{h, r, at}, now,
73	        Bool.and(String.eq(h, Lease.holder(l)), String.eq(r, Lease.renew_time(l))))
74	
75	# Whether another holder's lease is still live by our clock.
76	def Lease.held(+l: J.Json, +me: String, o: Obs, +now: U32, +dur: U32) -> Bool:
77	  match o:
78	    case Obs{h, r, +at}:
79	      Bool.and(Bool.not(String.is_empty(Lease.holder(l))),
80	        Bool.and(Bool.not(String.eq(Lease.holder(l), me)),
81	          U32.is_lt((now - at : U32), (dur * 1000 : U32))))
82	
83	def Lease.take.go(+l: J.Json, me: String, +dur: U32, +now: String, ours: Bool) -> J.Json:
84	  match ours:
85	    case True{}:
86	      J.Json.set(l, "spec", Lease.new_spec(me, dur,
87	        J.Json.str_or(Lease.spec(l, "acquireTime"), now), now, Lease.transitions(l)))
88	    case False{}:
89	      J.Json.set(l, "spec", Lease.new_spec(me, dur, now, now, (Lease.transitions(l) + 1 : U32)))
90	
91	# The Lease rewritten to name us, keeping its metadata (resourceVersion).
92	def Lease.take(+l: J.Json, +me: String, +dur: U32, +now: String) -> J.Json:
93	  Lease.take.go(l, me, dur, now, String.eq(Lease.holder(l), me))
94	
95	def Lease.decide.fin(+l: J.Json, +me: String, o: Obs, +dur: U32, +now: String, held: Bool) ->
96	  Decision:
97	  match held:
98	    case True{}:
99	      Decision{LWait{}, o}
100	    case False{}:
101	      Decision{LUpdate{Lease.take(l, me, dur, now)}, o}
102	
103	def Lease.decide.held(+l: J.Json, +me: String, +o: Obs, +now_ms: U32, +dur: U32, +now: String) ->
104	  Decision:
105	  Lease.decide.fin(l, me, o, dur, now, Lease.held(l, me, o, now_ms, dur))
106	
107	def Lease.decide.got(+l: J.Json, +me: String, o: Obs, +now_ms: U32, +dur: U32, +now: String) ->
108	  Decision:
109	  Lease.decide.held(l, me, Lease.observe(l, o, now_ms), now_ms, dur, now)
110	
111	# One round: the Lease read (None when absent), us, what we saw before,
112	# our clock in ms, the lease duration in seconds, and the time to stamp.
113	def Lease.decide(cur: Maybe<&2, J.Json>, ns: String, name: String, +me: String, o: Obs,
114	  +now_ms: U32, +dur: U32, +now: String) -> Decision:
115	  match cur:
116	    case None{}:
117	      Decision{LCreate{Lease.fresh(ns, name, me, dur, now)}, Obs{me, now, now_ms}}
118	    case Some{+l}:
119	      Lease.decide.got(l, me, o, now_ms, dur, now)
120	
121	# The Lease handed back: no holder, a one-second duration, so the next
122	# candidate takes it at once (client-go's ReleaseOnCancel).
123	def Lease.release(+l: J.Json, +now: String) -> J.Json:
124	  J.Json.set(l, "spec", Lease.new_spec("", 1, now, now, Lease.transitions(l)))