1	# Redis and Valkey client: RESP3 over TCP or TLS, with pipelining and a pool. Source: https://github.com/paymog/bend-kit/tree/main/redis
2	import Base
3	import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4	import bend-kit-wire@0.4.6.2/wire.bend as Wire
5	import bend-kit-dns@0.7.0.0/dns.bend as Dns
6	
7	# A RESP3 value (RESP2 replies decode into the same type). Text fields hold the raw
8	# bytes of the line. Dict (a RESP3 map) and Attr keep their pairs flat, key then value, as
9	# on the wire and in hiredis. Attr holds the attribute pairs, then the reply they annotate, last.
10	# Int is a signed 64-bit integer as two's-complement words.
11	type Val is Type:
12	  Simple{text: Bytes.Bytes}
13	  Error{text: Bytes.Bytes}
14	  Int{hi: U32, lo: U32}
15	  Bulk{data: Bytes.Bytes}
16	  Arr{items: List<&1, Val>}
17	  Null{}
18	  Boolean{value: Bool}
19	  Double{text: Bytes.Bytes}
20	  BigNum{text: Bytes.Bytes}
21	  BulkError{text: Bytes.Bytes}
22	  Verbatim{format: Bytes.Bytes, text: Bytes.Bytes}
23	  Dict{items: List<&1, Val>}
24	  Set{items: List<&1, Val>}
25	  Attr{items: List<&1, Val>}
26	  Push{items: List<&1, Val>}
27	
28	# Aggregate kinds, in the order of their type bytes below.
29	def kind.tag(+kind: U32) -> String:
30	  match kind:
31	    case 0:
32	      "*"
33	    case 1:
34	      "%"
35	    case 2:
36	      "~"
37	    case 3:
38	      "|"
39	    case _:
40	      ">"
41	
42	def kind.val(+kind: U32, xs: List<&1, Val>) -> Val:
43	  match kind:
44	    case 0:
45	      Arr{xs}
46	    case 1:
47	      Dict{xs}
48	    case 2:
49	      Set{xs}
50	    case 3:
51	      Attr{xs}
52	    case _:
53	      Push{xs}
54	
55	# ---- integers
56	
57	def i64.neg(+hi: U32, +lo: U32) -> U32 & U32:
58	  (((hi .^. 4294967295) + Bool.pick(U32, U32.is_eq(lo, 0), 1, 0) : U32), ((lo .^. 4294967295) + 1 : U32))
59	
60	# The 64-bit (hi, lo) divided by 10, and the remainder, over 16-bit limbs.
61	def u64.div10(+hi: U32, +lo: U32) -> U32 & U32 & U32:
62	  +a = (hi >> 16n : U32)
63	  +x = (((a % 10) << 16n) .|. (hi .&. 65535) : U32)
64	  +y = (((x % 10) << 16n) .|. (lo >> 16n) : U32)
65	  +z = (((y % 10) << 16n) .|. (lo .&. 65535) : U32)
66	  ((((a / 10) << 16n) .|. (x / 10) : U32), (((y / 10) << 16n) .|. (z / 10) : U32), (z % 10 : U32))
67	
68	def digit.put(z: Bool, +r: U32, acc: String) -> String:
69	  match z:
70	    case True{}:
71	      acc
72	    case False{}:
73	      SCon{Chr{(48 + r : U32)}, acc}
74	
75	# z is set once the value left to print is zero; 20 digits hold any u64.
76	def u64.show.go(f: Nat, q: U32 & U32 & U32, acc: String, +z: Bool) -> String:
77	  match f:
78	    case 0n:
79	      acc
80	    case 1n+p:
81	      # A + on a later tuple binder fails to check (https://github.com/bendlang/bend/issues/1077).
82	      (hi0, lo0, r) = q
83	      +hi = hi0
84	      +lo = lo0
85	      u64.show.go(p, u64.div10(hi, lo), digit.put(z, r, acc), Bool.or(z, U32.is_eq((hi .|. lo : U32), 0)))
86	
87	def u64.show(+hi: U32, +lo: U32) -> String:
88	  u64.show.go(20n, u64.div10(hi, lo), SNil{}, False{})
89	
90	def i64.show.of(r: U32 & U32, neg: Bool) -> String:
91	  (+hi, +lo) = r
92	  match neg:
93	    case True{}:
94	      "-" ++ u64.show(hi, lo)
95	    case False{}:
96	      u64.show(hi, lo)
97	
98	def i64.show(+hi: U32, +lo: U32) -> String:
99	  +neg = U32.is_le(2147483648, hi)
100	  i64.show.of(Bool.pick(U32 & U32, neg, i64.neg(hi, lo), (hi, lo)), neg)
101	
102	# ---- encoder
103	
104	# The output so far, whether every value had a RESP form, and the count of values written.
105	type E is Type:
106	  E{o: Bytes.Bytes, ok: Bool, n: U32}
107	
108	def fresh() -> E:
109	  E{Bytes.new(0), True{}, 0}
110	
111	def text(+s: String) -> Bytes.Bytes:
112	  Bytes.from_string(s)
113	
114	def none(m: Maybe<&2, U32>) -> Bool:
115	  match m:
116	    case None{}:
117	      True{}
118	    case Some{x}:
119	      False{}
120	
121	def clean.lf(+cr: Bool, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Bool:
122	  (b, m) = r
123	  (b, Bool.and(cr, none(m)))
124	
125	def clean.cr(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Bool:
126	  (b, m) = r
127	  clean.lf(none(m), Bytes.find(b, "\n"))
128	
129	# A line has no CR and no LF.
130	def clean(t: Bytes.Bytes) -> Bytes.Bytes & Bool:
131	  clean.cr(Bytes.find(t, "\r"))
132	
133	def put(st: E, +head: String, body: Bytes.Bytes, +good: Bool) -> E:
134	  E{o, +ok, +n} = st
135	  E{Bytes.append(Bytes.append(Bytes.append(o, text(head)), body), text("\r\n")), Bool.and(ok, good), (n + 1 : U32)}
136	
137	def line(st: E, +tag: String, r: Bytes.Bytes & Bool) -> E:
138	  (t, +good) = r
139	  put(st, tag, t, good)
140	
141	def blob(st: E, +tag: String, b: Bytes.Bytes) -> E:
142	  Bytes.Bytes{+len, buf} = b
143	  put(st, tag ++ U32.show(len) ++ "\r\n", Bytes.Bytes{len, buf}, True{})
144	
145	def verbatim.of(st: E, +tl: U32, r: Bytes.Bytes & U32, t: Bytes.Bytes) -> E:
146	  (f, +fl) = r
147	  put(st, "=" ++ U32.show((tl + 4 : U32)) ++ "\r\n", Bytes.append(Bytes.append(f, text(":")), t), U32.is_eq(fl, 3))
148	
149	def verbatim(st: E, f: Bytes.Bytes, t: Bytes.Bytes) -> E:
150	  Bytes.Bytes{+tl, tb} = t
151	  verbatim.of(st, tl, Bytes.length(f), Bytes.Bytes{tl, tb})
152	
153	# Maps count pairs and need an even count; Attr needs its pairs and one reply.
154	def agg.count(+kind: U32, +n: U32) -> U32:
155	  Bool.pick(U32, U32.is_eq(kind, 1), (n >> 1n : U32), Bool.pick(U32, U32.is_eq(kind, 3), (n >> 1n : U32), n))
156	
157	def agg.ok(+kind: U32, +n: U32) -> Bool:
158	  +odd = U32.is_eq((n .&. 1 : U32), 1)
159	  Bool.pick(Bool, U32.is_eq(kind, 1), Bool.not(odd), Bool.pick(Bool, U32.is_eq(kind, 3), odd, True{}))
160	
161	def agg(st: E, +kind: U32, sub: E) -> E:
162	  E{o, +ok, +n} = st
163	  E{body, +good, +k} = sub
164	  +head = kind.tag(kind) ++ U32.show(agg.count(kind, k)) ++ "\r\n"
165	  E{Bytes.append(Bytes.append(o, text(head)), body), Bool.and(ok, Bool.and(good, agg.ok(kind, k))), (n + 1 : U32)}
166	
167	def boolean(v: Bool) -> String:
168	  match v:
169	    case True{}:
170	      "t"
171	    case False{}:
172	      "f"
173	
174	def enc(xs: List<&1, Val>, st: E) -> E:
175	  match xs:
176	    case Nil{}:
177	      st
178	    case Con{Simple{t}, rest}:
179	      enc(rest, line(st, "+", clean(t)))
180	    case Con{Error{t}, rest}:
181	      enc(rest, line(st, "-", clean(t)))
182	    case Con{Int{+hi, +lo}, rest}:
183	      enc(rest, put(st, ":" ++ i64.show(hi, lo), Bytes.new(0), True{}))
184	    case Con{Bulk{d}, rest}:
185	      enc(rest, blob(st, "$", d))
186	    case Con{Arr{items}, rest}:
187	      enc(rest, agg(st, 0, enc(items, fresh())))
188	    case Con{Null{}, rest}:
189	      enc(rest, put(st, "_", Bytes.new(0), True{}))
190	    case Con{Boolean{v}, rest}:
191	      enc(rest, put(st, "#" ++ boolean(v), Bytes.new(0), True{}))
192	    case Con{Double{t}, rest}:
193	      enc(rest, line(st, ",", clean(t)))
194	    case Con{BigNum{t}, rest}:
195	      enc(rest, line(st, "(", clean(t)))
196	    case Con{BulkError{t}, rest}:
197	      enc(rest, blob(st, "!", t))
198	    case Con{Verbatim{f, t}, rest}:
199	      enc(rest, verbatim(st, f, t))
200	    case Con{Dict{items}, rest}:
201	      enc(rest, agg(st, 1, enc(items, fresh())))
202	    case Con{Set{items}, rest}:
203	      enc(rest, agg(st, 2, enc(items, fresh())))
204	    case Con{Attr{items}, rest}:
205	      enc(rest, agg(st, 3, enc(items, fresh())))
206	    case Con{Push{items}, rest}:
207	      enc(rest, agg(st, 4, enc(items, fresh())))
208	
209	def encode.done(st: E) -> Maybe<&1, Bytes.Bytes>:
210	  E{o, ok, n} = st
211	  match ok:
212	    case True{}:
213	      Some{o}
214	    case False{}:
215	      None{}
216	
217	# None when a line value holds CR or LF, a Dict has an odd item count, an Attr an even
218	# one, or a Verbatim format is not 3 bytes.
219	def encode(v: Val) -> Maybe<&1, Bytes.Bytes>:
220	  encode.done(enc([v], fresh()))
221	
222	def bulks(xs: List<&1, Bytes.Bytes>) -> List<&1, Val>:
223	  match xs:
224	    case Nil{}:
225	      Nil{}
226	    case Con{h, t}:
227	      Con{Bulk{h}, bulks(t)}
228	
229	def encode.command.done(st: E) -> Bytes.Bytes:
230	  E{o, ok, n} = st
231	  o
232	
233	# A command as a RESP array of bulk strings, the form every server accepts.
234	def encode.command(args: List<&1, Bytes.Bytes>) -> Bytes.Bytes:
235	  encode.command.done(enc([Arr{bulks(args)}], fresh()))
236	
237	# Byte-string arguments (one Char per octet) as Bytes.
238	def args(xs: List<&2, String>) -> List<&1, Bytes.Bytes>:
239	  match xs:
240	    case Nil{}:
241	      Nil{}
242	    case Con{h, t}:
243	      Con{Bytes.from_string(h), args(t)}
244	
245	# ---- decoder
246	
247	# A decimal line: sign, magnitude words, and whether it has 1 to 19 digits.
248	type Num is Data:
249	  Num{neg: Bool, hi: U32, lo: U32, ok: Bool}
250	
251	# n digits from i: (hi, lo) = (hi, lo) * 10 + d over 16-bit limbs. 19 digits stay below 2^64.
252	def num.go(n: Nat, r: Array<U32> & U32, +i: U32, +hi: U32, +lo: U32, +ok: Bool) -> Array<U32> & U32 & U32 & Bool:
253	  match n:
254	    case 0n:
255	      (a, c) = r
256	      (a, hi, lo, ok)
257	    case 1n+p:
258	      (a, +c) = r
259	      +d = (c - 48 : U32)
260	      +p0 = ((lo .&. 65535) * 10 + d : U32)
261	      +p1 = ((lo >> 16n) * 10 + (p0 >> 16n) : U32)
262	      num.go(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32), (hi * 10 + (p1 >> 16n) : U32), ((p1 << 16n) .|. (p0 .&. 65535) : U32), Bool.and(ok, U32.is_lt(d, 10)))
263	
264	def num.fin(+neg: Bool, +len: U32, r: Array<U32> & U32 & U32 & Bool) -> Bytes.Bytes & Num:
265	  (a, hi, lo, ok) = r
266	  (Bytes.Bytes{len, a}, Num{neg, hi, lo, ok})
267	
268	def num.digits(+neg: Bool, a: Array<U32>, +len: U32, +s: U32, +e: U32) -> Bytes.Bytes & Num:
269	  +k = (e - s : U32)
270	  num.fin(neg, len, num.go(U32.to_nat(k), Bytes.peek(a, s), s, 0, 0, Bool.and(U32.is_lt(0, k), U32.is_le(k, 19))))
271	
272	def num.sign(+len: U32, +s: U32, +e: U32, r: Array<U32> & U32) -> Bytes.Bytes & Num:
273	  (a, +c) = r
274	  +neg = Bool.and(U32.is_eq(c, 45), U32.is_lt(s, e))
275	  num.digits(neg, a, len, Bool.pick(U32, neg, (s + 1 : U32), s), e)
276	
277	# Bytes s until e, which are before len.
278	def num(b: Bytes.Bytes, +s: U32, +e: U32) -> Bytes.Bytes & Num:
279	  Bytes.Bytes{+len, buf} = b
280	  num.sign(len, s, e, Bytes.peek(buf, s))
281	
282	def int.mk(r: U32 & U32) -> Val:
283	  (hi, lo) = r
284	  Int{hi, lo}
285	
286	def int.val(good: Bool, +neg: Bool, +hi: U32, +lo: U32) -> Maybe<&1, Val>:
287	  match good:
288	    case True{}:
289	      Some{int.mk(Bool.pick(U32 & U32, neg, i64.neg(hi, lo), (hi, lo)))}
290	    case False{}:
291	      None{}
292	
293	# RESP integers are signed 64-bit: the magnitude may reach 2^63 only when negative.
294	def int.of(n: Num) -> Maybe<&1, Val>:
295	  Num{+neg, +hi, +lo, ok} = n
296	  +fits = Bool.or(U32.is_lt(hi, 2147483648), Bool.and(neg, Bool.and(U32.is_eq(hi, 2147483648), U32.is_eq(lo, 0))))
297	  int.val(Bool.and(ok, fits), neg, hi, lo)
298	
299	# A length or count below 2^31; -1 (RESP2 null) is 4294967295 and anything else is 4294967294.
300	def len.of(n: Num) -> U32:
301	  Num{+neg, +hi, +lo, +ok} = n
302	  +nil = Bool.and(ok, Bool.and(neg, Bool.and(U32.is_eq(hi, 0), U32.is_eq(lo, 1))))
303	  +good = Bool.and(ok, Bool.and(Bool.not(neg), Bool.and(U32.is_eq(hi, 0), U32.is_lt(lo, 2147483648))))
304	  Bool.pick(U32, good, lo, Bool.pick(U32, nil, 4294967295, 4294967294))
305	
306	# An open aggregate: its kind, the items still due, and the items so far, reversed.
307	type Frame is Type:
308	  Frame{kind: U32, left: U32, xs: List<&1, Val>}
309	
310	type St is Type:
311	  SHead{i: U32, stack: List<&1, Frame>}
312	  SDone{v: Val, i: U32, stack: List<&1, Frame>}
313	  SOk{v: Val, i: U32}
314	  SMore{}
315	  SBad{}
316	
317	# One reply and the index after it; More when the bytes end first; Bad when they cannot start a reply.
318	type Step is Type:
319	  Reply{v: Val, next: U32}
320	  More{}
321	  Bad{}
322	
323	def byte.of(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
324	  (b, m) = r
325	  match m:
326	    case Some{c}:
327	      (b, c)
328	    case None{}:
329	      (b, 256)
330	
331	# Byte i, or 256 past the end.
332	def byte(b: Bytes.Bytes, +i: U32) -> Bytes.Bytes & U32:
333	  byte.of(Bytes.get(b, i))
334	
335	def known(+t: U32) -> Bool:
336	  match t:
337	    case 43:
338	      True{}
339	    case 45:
340	      True{}
341	    case 58:
342	      True{}
343	    case 36:
344	      True{}
345	    case 42:
346	      True{}
347	    case 95:
348	      True{}
349	    case 35:
350	      True{}
351	    case 44:
352	      True{}
353	    case 40:
354	      True{}
355	    case 33:
356	      True{}
357	    case 61:
358	      True{}
359	    case 37:
360	      True{}
361	    case 126:
362	      True{}
363	    case 124:
364	      True{}
365	    case 62:
366	      True{}
367	    case _:
368	      False{}
369	
370	def null.st(ok: Bool, +next: U32, stack: List<&1, Frame>) -> St:
371	  match ok:
372	    case True{}:
373	      SDone{Null{}, next, stack}
374	    case False{}:
375	      SBad{}
376	
377	def line.val(+k: U32, t: Bytes.Bytes) -> Val:
378	  match k:
379	    case 0:
380	      Simple{t}
381	    case 1:
382	      Error{t}
383	    case 2:
384	      Double{t}
385	    case _:
386	      BigNum{t}
387	
388	def line.st(+k: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
389	  (b, t) = r
390	  (b, SDone{line.val(k, t), next, stack})
391	
392	def int.done(m: Maybe<&1, Val>, +next: U32, stack: List<&1, Frame>) -> St:
393	  match m:
394	    case Some{v}:
395	      SDone{v, next, stack}
396	    case None{}:
397	      SBad{}
398	
399	def int.st(+next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
400	  (b, n) = r
401	  (b, int.done(int.of(n), next, stack))
402	
403	def boolean.done(ok: Bool, +v: Bool, +next: U32, stack: List<&1, Frame>) -> St:
404	  match ok:
405	    case True{}:
406	      SDone{Boolean{v}, next, stack}
407	    case False{}:
408	      SBad{}
409	
410	def boolean.st(+next: U32, stack: List<&1, Frame>, +one: Bool, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
411	  (b, +c) = r
412	  (b, boolean.done(Bool.and(one, Bool.or(U32.is_eq(c, 116), U32.is_eq(c, 102))), U32.is_eq(c, 116), next, stack))
413	
414	def blob.mk(+kind: U32, d: Bytes.Bytes) -> Val:
415	  match kind:
416	    case 0:
417	      Bulk{d}
418	    case _:
419	      BulkError{d}
420	
421	def blob.done(+kind: U32, +after: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
422	  (b, d) = r
423	  (b, SDone{blob.mk(kind, d), after, stack})
424	
425	def verb.text(f: Bytes.Bytes, +after: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
426	  (b, t) = r
427	  (b, SDone{Verbatim{f, t}, after, stack})
428	
429	def verb.fmt(+n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
430	  (b, f) = r
431	  verb.text(f, (next + n + 2 : U32), stack, Bytes.slice(b, (next + 4 : U32), (n - 4 : U32)))
432	
433	def verb.ok(ok: Bool, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
434	  match ok:
435	    case True{}:
436	      verb.fmt(n, next, stack, Bytes.slice(b, next, 3))
437	    case False{}:
438	      (b, SBad{})
439	
440	def verb.colon(+n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
441	  (b, +c) = r
442	  verb.ok(U32.is_eq(c, 58), n, next, stack, b)
443	
444	# A verbatim string is a 3-byte format, a colon, then the text.
445	def verb.len(ok: Bool, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
446	  match ok:
447	    case True{}:
448	      verb.colon(n, next, stack, byte(b, (next + 3 : U32)))
449	    case False{}:
450	      (b, SBad{})
451	
452	def blob.val(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
453	  match kind:
454	    case 2:
455	      verb.len(U32.is_le(4, n), n, next, stack, b)
456	    case _:
457	      blob.done(kind, (next + n + 2 : U32), stack, Bytes.slice(b, next, n))
458	
459	def blob.crlf(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bool) -> Bytes.Bytes & St:
460	  (b, ok) = r
461	  match ok:
462	    case True{}:
463	      blob.val(kind, n, next, stack, b)
464	    case False{}:
465	      (b, SBad{})
466	
467	def crlf.at(b: Bytes.Bytes, +j: U32) -> Bytes.Bytes & Bool:
468	  Bytes.Bytes{+len, buf} = b
469	  Bytes.at.fin(len, Bytes.at(buf, "\r\n", j))
470	
471	def blob.fit(ok: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
472	  match ok:
473	    case True{}:
474	      blob.crlf(kind, n, next, stack, crlf.at(b, (next + n : U32)))
475	    case False{}:
476	      (b, SMore{})
477	
478	def blob.room(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
479	  (b, +len) = r
480	  blob.fit(Bytes.fits(len, next, (n + 2 : U32)), kind, n, next, stack, b)
481	
482	def blob.bad(bad: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
483	  match bad:
484	    case True{}:
485	      (b, SBad{})
486	    case False{}:
487	      blob.room(kind, n, next, stack, Bytes.length(b))
488	
489	# $-1 is RESP2's null bulk string; ! and = have no null form.
490	def blob.nil(nil: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
491	  match nil:
492	    case True{}:
493	      (b, null.st(U32.is_eq(kind, 0), next, stack))
494	    case False{}:
495	      blob.bad(U32.is_eq(n, 4294967294), kind, n, next, stack, b)
496	
497	# kind 0 is $, 1 is !, 2 is =.
498	def blob.st(+kind: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
499	  (b, n) = r
500	  +k = len.of(n)
501	  blob.nil(U32.is_eq(k, 4294967295), kind, k, next, stack, b)
502	
503	def agg.open(empty: Bool, +kind: U32, +left: U32, +next: U32, stack: List<&1, Frame>) -> St:
504	  match empty:
505	    case True{}:
506	      SDone{kind.val(kind, Nil{}), next, stack}
507	    case False{}:
508	      SHead{next, Con{Frame{kind, left, Nil{}}, stack}}
509	
510	# A map of n pairs has 2n items; an attribute adds the reply after its pairs.
511	def agg.left(+kind: U32, +n: U32) -> U32:
512	  Bool.pick(U32, U32.is_eq(kind, 1), (n * 2 : U32), Bool.pick(U32, U32.is_eq(kind, 3), (n * 2 + 1 : U32), n))
513	
514	def agg.bad(bad: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>) -> St:
515	  match bad:
516	    case True{}:
517	      SBad{}
518	    case False{}:
519	      +left = agg.left(kind, n)
520	      agg.open(U32.is_eq(left, 0), kind, left, next, stack)
521	
522	# *-1 is RESP2's null array.
523	def agg.nil(nil: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>) -> St:
524	  match nil:
525	    case True{}:
526	      null.st(U32.is_eq(kind, 0), next, stack)
527	    case False{}:
528	      agg.bad(U32.is_eq(n, 4294967294), kind, n, next, stack)
529	
530	def agg.st(+kind: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
531	  (b, n) = r
532	  +k = len.of(n)
533	  (b, agg.nil(U32.is_eq(k, 4294967295), kind, k, next, stack))
534	
535	# The type byte t, then the line from s until the CRLF at e.
536	def item(+t: U32, +s: U32, +e: U32, b: Bytes.Bytes, stack: List<&1, Frame>) -> Bytes.Bytes & St:
537	  match t:
538	    case 43:
539	      line.st(0, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
540	    case 45:
541	      line.st(1, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
542	    case 44:
543	      line.st(2, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
544	    case 40:
545	      line.st(3, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
546	    case 58:
547	      int.st((e + 2 : U32), stack, num(b, s, e))
548	    case 95:
549	      (b, null.st(U32.is_eq(s, e), (e + 2 : U32), stack))
550	    case 35:
551	      boolean.st((e + 2 : U32), stack, U32.is_eq(e, (s + 1 : U32)), byte(b, s))
552	    case 36:
553	      blob.st(0, (e + 2 : U32), stack, num(b, s, e))
554	    case 33:
555	      blob.st(1, (e + 2 : U32), stack, num(b, s, e))
556	    case 61:
557	      blob.st(2, (e + 2 : U32), stack, num(b, s, e))
558	    case 42:
559	      agg.st(0, (e + 2 : U32), stack, num(b, s, e))
560	    case 37:
561	      agg.st(1, (e + 2 : U32), stack, num(b, s, e))
562	    case 126:
563	      agg.st(2, (e + 2 : U32), stack, num(b, s, e))
564	    case 124:
565	      agg.st(3, (e + 2 : U32), stack, num(b, s, e))
566	    case _:
567	      agg.st(4, (e + 2 : U32), stack, num(b, s, e))
568	
569	def head.eol(+t: U32, +i: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & St:
570	  (b, m) = r
571	  match m:
572	    case Some{+e}:
573	      item(t, (i + 1 : U32), e, b, stack)
574	    case None{}:
575	      (b, SMore{})
576	
577	def head.end(end: Bool) -> St:
578	  match end:
579	    case True{}:
580	      SMore{}
581	    case False{}:
582	      SBad{}
583	
584	def head.known(ok: Bool, +t: U32, +i: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
585	  match ok:
586	    case True{}:
587	      head.eol(t, i, stack, Bytes.find.from(b, "\r\n", (i + 1 : U32)))
588	    case False{}:
589	      (b, head.end(U32.is_eq(t, 256)))
590	
591	def head(+i: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
592	  (b, +t) = r
593	  head.known(known(t), t, i, stack, b)
594	
595	def done.last(last: Bool, +kind: U32, +left: U32, xs: List<&1, Val>, +i: U32, rest: List<&1, Frame>) -> St:
596	  match last:
597	    case True{}:
598	      SDone{kind.val(kind, List.reverse(&1, Val, xs)), i, rest}
599	    case False{}:
600	      SHead{i, Con{Frame{kind, (left - 1 : U32), xs}, rest}}
601	
602	def done(stack: List<&1, Frame>, v: Val, +i: U32) -> St:
603	  match stack:
604	    case Nil{}:
605	      SOk{v, i}
606	    case Con{fr, rest}:
607	      Frame{+kind, +left, xs} = fr
608	      done.last(U32.is_eq(left, 1), kind, left, Con{v, xs}, i, rest)
609	
610	def run(fuel: Nat, r: Bytes.Bytes & St) -> Bytes.Bytes & Step:
611	  match fuel:
612	    case 0n:
613	      (b, st) = r
614	      (b, Bad{})
615	    case 1n+f:
616	      (b, st) = r
617	      match st:
618	        case SOk{v, +i}:
619	          (b, Reply{v, i})
620	        case SMore{}:
621	          (b, More{})
622	        case SBad{}:
623	          (b, Bad{})
624	        case SHead{+i, stack}:
625	          run(f, head(i, stack, byte(b, i)))
626	        case SDone{v, +i, stack}:
627	          run(f, (b, done(stack, v, i)))
628	
629	# One reply starting at byte i of b. Every item takes at least 3 bytes and each step
630	# reads an item or closes one, so 2 * (len - i) + 2 steps are enough.
631	def decode(b: Bytes.Bytes, +i: U32) -> Bytes.Bytes & Step:
632	  Bytes.Bytes{+len, buf} = b
633	  run(Nat.add(Nat.mul(U32.to_nat((len - i : U32)), 2n), 2n), (Bytes.Bytes{len, buf}, SHead{i, Nil{}}))
634	
635	# ---- incremental reader
636	
637	# Bytes read from the server, and the index of the first byte not yet decoded.
638	type Reader is Type:
639	  Reader{buf: Bytes.Bytes, pos: U32}
640	
641	def reader() -> Reader:
642	  Reader{Bytes.new(0), 0}
643	
644	def feed.at(zero: Bool, buf: Bytes.Bytes, +pos: U32, chunk: Bytes.Bytes) -> Reader:
645	  match zero:
646	    case True{}:
647	      Reader{Bytes.append(buf, chunk), 0}
648	    case False{}:
649	      Reader{Bytes.append(Bytes.snd(Bytes.slice(buf, pos, 4294967295)), chunk), 0}
650	
651	# Drops the decoded bytes, then adds chunk.
652	# ponytail: a reply split over k reads is decoded from its start k times; keep state per item if huge arrays arrive slowly.
653	def feed(r: Reader, chunk: Bytes.Bytes) -> Reader:
654	  Reader{buf, +pos} = r
655	  feed.at(U32.is_zero(pos), buf, pos, chunk)
656	
657	type Next is Type:
658	  Got{v: Val}
659	  Want{}
660	  Broken{}
661	
662	def next.of(+pos: U32, r: Bytes.Bytes & Step) -> Reader & Next:
663	  (b, s) = r
664	  match s:
665	    case Reply{v, +i}:
666	      (Reader{b, i}, Got{v})
667	    case More{}:
668	      (Reader{b, pos}, Want{})
669	    case Bad{}:
670	      (Reader{b, pos}, Broken{})
671	
672	# The next whole reply, Want when it has not all arrived, or Broken for bytes that are not RESP.
673	def next(r: Reader) -> Reader & Next:
674	  Reader{buf, +pos} = r
675	  next.of(pos, decode(buf, pos))
676	
677	def pending.of(+pos: U32, r: Bytes.Bytes & U32) -> Reader & U32:
678	  (b, +len) = r
679	  (Reader{b, pos}, (len - pos : U32))
680	
681	# The count of bytes read but not yet decoded.
682	def pending(r: Reader) -> Reader & U32:
683	  Reader{buf, +pos} = r
684	  pending.of(pos, Bytes.length(buf))
685	
686	# ---- connections
687	
688	# host is a name or a numeric address; TLS checks the certificate against it. Empty user
689	# and pass send no AUTH (a pass alone authenticates as "default"); db 0 sends no SELECT.
690	# ms bounds each connect and read (0: none).
691	type Config is Data:
692	  Config{host: String, port: U32, tls: Bool, user: String, pass: String, db: U32, ms: U32}
693	
694	def config(+host: String, +port: U32) -> Config:
695	  Config{host, port, False{}, "", "", 0, 10000}
696	
697	# A failed call closes its connection. Rejected carries the server's error reply to the handshake.
698	type Failure is Type:
699	  IoFail{code: U32, why: String}
700	  Closed{}
701	  Malformed{}
702	  Rejected{reply: Val}
703	
704	# key names the pool bucket: TLS, user, host, port, and db.
705	type Conn is Type:
706	  Conn{tls: Bool, sock: Socket, rd: Reader, key: String, ms: U32}
707	
708	def io.close(tls: Bool, s: Socket) -> IO(Unit):
709	  match tls:
710	    case True{}:
711	      Wire.tls.close(s)
712	    case False{}:
713	      Socket.close(s)
714	
715	def io.send(tls: Bool, s: Socket, +len: U32, buf: Array<U32>) -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
716	  match tls:
717	    case True{}:
718	      Wire.tls.send.words(s, len, buf)
719	    case False{}:
720	      Wire.send.words(s, len, buf)
721	
722	def io.recv(tls: Bool, s: Socket, +ms: U32) -> IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
723	  match tls:
724	    case True{}:
725	      Wire.tls.recv.words(s, 65536, ms)
726	    case False{}:
727	      Wire.recv.words(s, 65536, ms)
728	
729	def close(c: Conn) -> IO(Unit):
730	  Conn{tls, sock, rd, key, ms} = c
731	  io.close(tls, sock)
732	
733	def send.after(+tls: Bool, rd: Reader, +key: String, +ms: U32, m: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, Failure, Conn>):
734	  (s, r) = m
735	  match r:
736	    case Fail{(code, why)}:
737	      do IO<Result<&1, &1, Failure, Conn>>:
738	        io.close(tls, s)
739	        return Fail{IoFail{code, why}}
740	    case Done{u}:
741	      IO.pure(Result<&1, &1, Failure, Conn>, Done{Conn{tls, s, rd, key, ms}})
742	
743	def send(c: Conn, data: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn>):
744	  Conn{+tls, sock, rd, +key, +ms} = c
745	  Bytes.Bytes{+len, buf} = data
746	  do IO<Result<&1, &1, Failure, Conn>>:
747	    m : Socket & Result<&1, &1, U32 & String, Unit> <- io.send(tls, sock, len, buf)
748	    send.after(tls, rd, key, ms, m)
749	
750	# RNext holds the reader after its last decode attempt; RFail a socket to close.
751	type Rd is Type:
752	  RNext{tls: Bool, sock: Socket, key: String, ms: U32, r: Reader & Next}
753	  RFail{tls: Bool, sock: Socket, why: Failure}
754	
755	def got.len(zero: Bool, +tls: Bool, +key: String, +ms: U32, rd: Reader, s: Socket, +len: U32, words: Array<U32>) -> Rd:
756	  match zero:
757	    case True{}:
758	      RFail{tls, s, Closed{}}
759	    case False{}:
760	      RNext{tls, s, key, ms, next(feed(rd, Bytes.Bytes{len, words}))}
761	
762	def got(+tls: Bool, +key: String, +ms: U32, rd: Reader, m: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>) -> Rd:
763	  (s, r) = m
764	  match r:
765	    case Fail{(code, why)}:
766	      RFail{tls, s, IoFail{code, why}}
767	    case Done{(+len, words)}:
768	      got.len(U32.is_zero(len), tls, key, ms, rd, s, len, words)
769	
770	def read.fail(tls: Bool, sock: Socket, why: Failure) -> IO(Result<&1, &1, Failure, Conn & Val>):
771	  do IO<Result<&1, &1, Failure, Conn & Val>>:
772	    io.close(tls, sock)
773	    return Fail{why}
774	
775	# One recv per step. 2^24 reads of up to 64 KiB pass Redis's 512 MiB bulk limit.
776	def read.loop(fuel: Nat, st: Rd) -> IO(Result<&1, &1, Failure, Conn & Val>):
777	  match fuel:
778	    case 0n:
779	      match st:
780	        case RNext{tls, sock, key, ms, r}:
781	          read.fail(tls, sock, Malformed{})
782	        case RFail{tls, sock, why}:
783	          read.fail(tls, sock, why)
784	    case 1n+f:
785	      match st:
786	        case RFail{tls, sock, why}:
787	          read.fail(tls, sock, why)
788	        case RNext{+tls, sock, +key, +ms, r}:
789	          (rd, nx) = r
790	          match nx:
791	            case Got{v}:
792	              IO.pure(Result<&1, &1, Failure, Conn & Val>, Done{(Conn{tls, sock, rd, key, ms}, v)})
793	            case Broken{}:
794	              read.fail(tls, sock, Malformed{})
795	            case Want{}:
796	              do IO<Result<&1, &1, Failure, Conn & Val>>:
797	                m : Socket & Result<&1, &1, U32 & String, U32 & Array<U32>> <- io.recv(tls, sock, ms)
798	                read.loop(f, got(tls, key, ms, rd, m))
799	
800	# The next reply. Bytes already read are decoded before the socket is asked for more.
801	def read(c: Conn) -> IO(Result<&1, &1, Failure, Conn & Val>):
802	  Conn{tls, sock, rd, key, ms} = c
803	  read.loop(Nat.mul(65536n, 256n), RNext{tls, sock, key, ms, next(rd)})
804	
805	def read.push(xs: List<&1, Val>, r: Result<&1, &1, Failure, Conn & Val>) -> Result<&1, &1, Failure, Conn & List<&1, Val>>:
806	  match r:
807	    case Fail{e}:
808	      Fail{e}
809	    case Done{(c, v)}:
810	      Done{(c, Con{v, xs})}
811	
812	# n more replies onto the reversed list in st.
813	def read.n(n: Nat, st: Result<&1, &1, Failure, Conn & List<&1, Val>>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
814	  match n:
815	    case 0n:
816	      match st:
817	        case Fail{e}:
818	          IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
819	        case Done{(c, xs)}:
820	          IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Done{(c, List.reverse(&1, Val, xs))})
821	    case 1n+p:
822	      match st:
823	        case Fail{e}:
824	          IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
825	        case Done{(c, xs)}:
826	          do IO<Result<&1, &1, Failure, Conn & List<&1, Val>>>:
827	            r : Result<&1, &1, Failure, Conn & Val> <- read(c)
828	            read.n(p, read.push(xs, r))
829	
830	def commands(cmds: List<&1, List<&1, Bytes.Bytes>>, o: Bytes.Bytes, +n: Nat) -> Bytes.Bytes & Nat:
831	  match cmds:
832	    case Nil{}:
833	      (o, n)
834	    case Con{h, t}:
835	      commands(t, Bytes.append(o, encode.command(h)), (1n + n : Nat))
836	
837	def pipeline.sent(+n: Nat, r: Result<&1, &1, Failure, Conn>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
838	  match r:
839	    case Fail{e}:
840	      IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
841	    case Done{c}:
842	      read.n(n, Done{(c, Nil{})})
843	
844	def pipeline.of(c: Conn, r: Bytes.Bytes & Nat) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
845	  (data, +n) = r
846	  do IO<Result<&1, &1, Failure, Conn & List<&1, Val>>>:
847	    s : Result<&1, &1, Failure, Conn> <- send(c, data)
848	    pipeline.sent(n, s)
849	
850	# Sends every command in one write, then reads one reply per command, in order.
851	# Error replies come back as Error or BulkError values, not as failures.
852	def pipeline(c: Conn, cmds: List<&1, List<&1, Bytes.Bytes>>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
853	  pipeline.of(c, commands(cmds, Bytes.new(0), 0n))
854	
855	def command.sent(r: Result<&1, &1, Failure, Conn>) -> IO(Result<&1, &1, Failure, Conn & Val>):
856	  match r:
857	    case Fail{e}:
858	      IO.pure(Result<&1, &1, Failure, Conn & Val>, Fail{e})
859	    case Done{c}:
860	      read(c)
861	
862	# One command and its reply, for example command(c, args(["SET", "k", "v"])).
863	def command(c: Conn, argv: List<&1, Bytes.Bytes>) -> IO(Result<&1, &1, Failure, Conn & Val>):
864	  do IO<Result<&1, &1, Failure, Conn & Val>>:
865	    r : Result<&1, &1, Failure, Conn> <- send(c, encode.command(argv))
866	    command.sent(r)
867	
868	def get(c: Conn, key: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
869	  command(c, [text("GET"), key])
870	
871	def set(c: Conn, key: Bytes.Bytes, value: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
872	  command(c, [text("SET"), key, value])
873	
874	def incr(c: Conn, key: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
875	  command(c, [text("INCR"), key])
876	
877	def first.err(xs: List<&1, Val>) -> Maybe<&1, Val>:
878	  match xs:
879	    case Nil{}:
880	      None{}
881	    case Con{Error{t}, rest}:
882	      Some{Error{t}}
883	    case Con{BulkError{t}, rest}:
884	      Some{BulkError{t}}
885	    case Con{v, rest}:
886	      first.err(rest)
887	
888	def hello.check(c: Conn, e: Maybe<&1, Val>) -> IO(Result<&1, &1, Failure, Conn>):
889	  match e:
890	    case None{}:
891	      IO.pure(Result<&1, &1, Failure, Conn>, Done{c})
892	    case Some{v}:
893	      do IO<Result<&1, &1, Failure, Conn>>:
894	        close(c)
895	        return Fail{Rejected{v}}
896	
897	def hello.got(r: Result<&1, &1, Failure, Conn & List<&1, Val>>) -> IO(Result<&1, &1, Failure, Conn>):
898	  match r:
899	    case Fail{e}:
900	      IO.pure(Result<&1, &1, Failure, Conn>, Fail{e})
901	    case Done{(c, xs)}:
902	      hello.check(c, first.err(xs))
903	
904	def hello.auth(+user: String, +pass: String) -> List<&2, String>:
905	  Bool.pick(List<&2, String>, String.eq(pass, ""), Nil{}, ["AUTH", Bool.pick(String, String.eq(user, ""), "default", user), pass])
906	
907	def hello.select(+db: U32) -> List<&1, List<&1, Bytes.Bytes>>:
908	  Bool.pick(List<&1, List<&1, Bytes.Bytes>>, U32.is_zero(db), Nil{}, [args(["SELECT", U32.show(db)])])
909	
910	# HELLO 3 (with AUTH), then SELECT, in one pipeline.
911	def hello(c: Conn, +user: String, +pass: String, +db: U32) -> IO(Result<&1, &1, Failure, Conn>):
912	  do IO<Result<&1, &1, Failure, Conn>>:
913	    r : Result<&1, &1, Failure, Conn & List<&1, Val>> <- pipeline(c, Con{args(List.append(&2, String, ["HELLO", "3"], hello.auth(user, pass))), hello.select(db)})
914	    hello.got(r)
915	
916	def key(cfg: Config) -> String:
917	  Config{+host, +port, +tls, +user, pass, +db, ms} = cfg
918	  Bool.pick(String, tls, "rediss://", "redis://") ++ user ++ "@" ++ host ++ ":" ++ U32.show(port) ++ "/" ++ U32.show(db)
919	
920	def tls.done(+key: String, +ms: U32, +user: String, +pass: String, +db: U32, m: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, Failure, Conn>):
921	  (s, r) = m
922	  match r:
923	    case Fail{(code, why)}:
924	      do IO<Result<&1, &1, Failure, Conn>>:
925	        Wire.tls.close(s)
926	        return Fail{IoFail{code, why}}
927	    case Done{u}:
928	      hello(Conn{True{}, s, reader(), key, ms}, user, pass, db)
929	
930	def dialed(r: Result<&1, &1, U32 & String, Socket>, tls: Bool, +k: String, +host: String, +user: String, +pass: String, +db: U32, +ms: U32) -> IO(Result<&1, &1, Failure, Conn>):
931	  match r:
932	    case Fail{(code, why)}:
933	      IO.pure(Result<&1, &1, Failure, Conn>, Fail{IoFail{code, why}})
934	    case Done{s}:
935	      match tls:
936	        case True{}:
937	          do IO<Result<&1, &1, Failure, Conn>>:
938	            m : Socket & Result<&1, &1, U32 & String, Unit> <- Wire.tls.connect(s, host, ms)
939	            tls.done(k, ms, user, pass, db, m)
940	        case False{}:
941	          hello(Conn{False{}, s, reader(), k, ms}, user, pass, db)
942	
943	def dial.after(r: Result<&1, &1, U32 & String, Socket>, +ip: String, +port: U32, +ms: U32) -> IO(Result<&1, &1, U32 & String, Socket>):
944	  match r:
945	    case Done{s}:
946	      IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
947	    case Fail{e}:
948	      Wire.connect(ip, port, ms)
949	
950	def dial.more(prev: IO(Result<&1, &1, U32 & String, Socket>), +ip: String, +port: U32, +ms: U32) -> IO(Result<&1, &1, U32 & String, Socket>):
951	  do IO<Result<&1, &1, U32 & String, Socket>>:
952	    r : Result<&1, &1, U32 & String, Socket> <- prev
953	    dial.after(r, ip, port, ms)
954	
955	# Each address in turn, until one connects.
956	def dial(ips: List<&2, String>, +port: U32, +ms: U32, acc: IO(Result<&1, &1, U32 & String, Socket>)) -> IO(Result<&1, &1, U32 & String, Socket>):
957	  match ips:
958	    case Nil{}:
959	      acc
960	    case Con{ip, t}:
961	      dial(t, port, ms, dial.more(acc, ip, port, ms))
962	
963	def connect.to(+k: String, cfg: Config) -> IO(Result<&1, &1, Failure, Conn>):
964	  Config{+host, +port, +tls, +user, +pass, +db, +ms} = cfg
965	  do IO<Result<&1, &1, Failure, Conn>>:
966	    ips : List<&2, String> <- Dns.resolve.all(host)
967	    r : Result<&1, &1, U32 & String, Socket> <- dial(ips, port, ms, IO.pure(Result<&1, &1, U32 & String, Socket>, Fail{(0, "no address for " ++ host)}))
968	    dialed(r, tls, k, host, user, pass, db, ms)
969	
970	# TCP (and TLS, which checks the certificate and host name), then the handshake.
971	def connect(+cfg: Config) -> IO(Result<&1, &1, Failure, Conn>):
972	  connect.to(key(cfg), cfg)
973	
974	# ---- pool
975	
976	# Idle connections per key, up to cap per key.
977	type Pool is Type:
978	  Pool{cap: U32, idle: Map<&1, List<&1, Conn>>}
979	
980	def pool.new.with(+cap: U32) -> Pool:
981	  Pool{cap, Map.new(&1, List<&1, Conn>)}
982	
983	def pool.new() -> Pool:
984	  pool.new.with(8)
985	
986	def pool.take(+cap: U32, +k: String, r: Map<&1, List<&1, Conn>> & Maybe<&1, List<&1, Conn>>) -> Pool & Maybe<&1, Conn>:
987	  (m, got) = r
988	  match got:
989	    case None{}:
990	      (Pool{cap, m}, None{})
991	    case Some{xs}:
992	      match xs:
993	        case Nil{}:
994	          (Pool{cap, m}, None{})
995	        case Con{c, t}:
996	          (Pool{cap, Map.set(&1, List<&1, Conn>, m, k, t)}, Some{c})
997	
998	def pool.opened(p: Pool, r: Result<&1, &1, Failure, Conn>) -> IO(Pool & Result<&1, &1, Failure, Conn>):
999	  IO.pure(Pool & Result<&1, &1, Failure, Conn>, (p, r))
1000	
1001	def pool.got(cfg: Config, r: Pool & Maybe<&1, Conn>) -> IO(Pool & Result<&1, &1, Failure, Conn>):
1002	  (p, m) = r
1003	  match m:
1004	    case Some{c}:
1005	      IO.pure(Pool & Result<&1, &1, Failure, Conn>, (p, Done{c}))
1006	    case None{}:
1007	      do IO<Pool & Result<&1, &1, Failure, Conn>>:
1008	        c : Result<&1, &1, Failure, Conn> <- connect(cfg)
1009	        pool.opened(p, c)
1010	
1011	# The idle connection given back last, or a new one.
1012	# ponytail: an idle connection the server has since closed fails on its first command; add a PING check if that bites.
1013	def pool.get(p: Pool, +cfg: Config) -> IO(Pool & Result<&1, &1, Failure, Conn>):
1014	  Pool{+cap, idle} = p
1015	  +k = key(cfg)
1016	  pool.got(cfg, pool.take(cap, k, Map.pop(&1, List<&1, Conn>, idle, k)))
1017	
1018	# The first n connections; the rest are closed.
1019	def conns.cut(xs: List<&1, Conn>, n: Nat) -> IO(List<&1, Conn>):
1020	  match xs:
1021	    case Nil{}:
1022	      IO.pure(List<&1, Conn>, Nil{})
1023	    case Con{c, t}:
1024	      match n:
1025	        case 0n:
1026	          do IO<List<&1, Conn>>:
1027	            close(c)
1028	            conns.cut(t, 0n)
1029	        case 1n+k:
1030	          do IO<List<&1, Conn>>:
1031	            r : List<&1, Conn> <- conns.cut(t, k)
1032	            return c <> r
1033	
1034	def pool.keep(+cap: U32, +k: String, c: Conn, r: Map<&1, List<&1, Conn>> & Maybe<&1, List<&1, Conn>>) -> IO(Pool):
1035	  (m, old) = r
1036	  do IO<Pool>:
1037	    xs : List<&1, Conn> <- conns.cut(c <> Maybe.default(&1, List<&1, Conn>, old, Nil{}), U32.to_nat(cap))
1038	    return Pool{cap, Map.set(&1, List<&1, Conn>, m, k, xs)}
1039	
1040	def pool.give(p: Pool, c: Conn) -> IO(Pool):
1041	  Pool{+cap, idle} = p
1042	  Conn{tls, sock, rd, +k, ms} = c
1043	  pool.keep(cap, k, Conn{tls, sock, rd, k, ms}, Map.pop(&1, List<&1, Conn>, idle, k))
1044	
1045	def pool.clean(clean: Bool, p: Pool, c: Conn) -> IO(Pool):
1046	  match clean:
1047	    case True{}:
1048	      pool.give(p, c)
1049	    case False{}:
1050	      do IO<Pool>:
1051	        close(c)
1052	        return p
1053	
1054	def pool.left(p: Pool, +tls: Bool, sock: Socket, +key: String, +ms: U32, r: Reader & U32) -> IO(Pool):
1055	  (rd, +n) = r
1056	  pool.clean(U32.is_zero(n), p, Conn{tls, sock, rd, key, ms})
1057	
1058	# Gives c back for reuse. A connection with unread reply bytes is out of step, so it is closed.
1059	def pool.put(p: Pool, c: Conn) -> IO(Pool):
1060	  Conn{+tls, sock, rd, +key, +ms} = c
1061	  pool.left(p, tls, sock, key, ms, pending(rd))
1062	
1063	def pool.close.go(xs: List<&1, List<&1, Conn>>) -> IO(Unit):
1064	  match xs:
1065	    case Nil{}:
1066	      IO.pure(Unit, Unit{})
1067	    case Con{cs, t}:
1068	      do IO<Unit>:
1069	        none : List<&1, Conn> <- conns.cut(cs, 0n)
1070	        pool.close.go(t)
1071	
1072	def pool.close(p: Pool) -> IO(Unit):
1073	  Pool{cap, idle} = p
1074	  pool.close.go(Map.values(&1, List<&1, Conn>, idle))