def Leader.run

bend-kube-runtime@0.2.0.0 · src/leader.bend:96

import Base
import bend-kube-runtime@0.2.0.0/src/json.bend as J
import bend-kube-runtime@0.2.0.0/src/config.bend as K
import bend-kube-runtime@0.2.0.0/src/client.bend as Cl
import bend-kube-runtime@0.2.0.0/src/time.bend as T
import bend-kube-runtime@0.2.0.0/src/sys.bend as Y
import bend-kube-runtime@0.2.0.0/src/log.bend as L
import bend-kube-runtime@0.2.0.0/src/lease.bend as E

96# The election loop. start runs once, when the lease is first ours.
97def 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))