src/coordinator/Client.bend
mylsm-lsm-store@0.6.0.0 · 478 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.0/src/coordinator/ClientPolicy.bend as Policy
4import mylsm-lsm-store@0.6.0.0/src/coordinator/SessionPolicy.bend as SessionPolicy
5import mylsm-lsm-store@0.6.0.0/src/coordinator/Io.bend as Io
6import mylsm-lsm-store@0.6.0.0/src/coordinator/Operation.bend as Operation
7import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
8import mylsm-lsm-store@0.6.0.0/src/coordinator/UnixSocketIo.bend as UnixSocket
9import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
10
11# Returns connection, transport, correlation, and retry-boundary outcomes.
12type Error is Type:
13 ConnectFailure{error: UnixSocket.Error}
14 RetryRead{error: Io.Error}
15 CommitUnknown{error: Io.Error}
16 FrameTooLarge{}
17 StaleOwner{}
18 CorrelationMismatch{}
19 MalformedResponse{}
20 RequestIdsExhausted{}
21 OwnerRejected{status: Protocol.ResponseStatus}
22
23# Keeps one canonical endpoint, owner generation, and request sequence.
24type Session is Type:
25 Session{endpoint: String, generation: U32,
26 requests: Policy.RequestSequence}
27
28def u32.clone.pair.from(triple: U32 & U32 & U32) -> U32 & U32:
29 match triple:
30 case (first, (second, _)): (first, second)
31
32def u32.clone.pair(value: U32) -> U32 & U32:
33 u32.clone.pair.from(Protocol.u32.duplicate.triple(value))
34
35def request.clone.payload(
36 versions: U32 & U32,
37 generations: U32 & U32,
38 request_ids: U32 & U32,
39 operations: U32 & U32,
40 payloads: Bytes.Bytes & Bytes.Bytes
41) -> Protocol.Request & Protocol.Request:
42 match versions:
43 case (version, version_copy):
44 match generations:
45 case (generation, generation_copy):
46 match request_ids:
47 case (request_id, request_id_copy):
48 match operations:
49 case (operation, operation_copy):
50 match payloads:
51 case (first, second):
52 (Protocol.Request{version, generation, request_id,
53 operation, first},
54 Protocol.Request{version_copy, generation_copy,
55 request_id_copy, operation_copy, second})
56
57def request.clone.pair(
58 request: Protocol.Request
59) -> Protocol.Request & Protocol.Request:
60 match request:
61 case Protocol.Request{version, generation, request_id, operation, payload}:
62 request.clone.payload(u32.clone.pair(version),
63 u32.clone.pair(generation), u32.clone.pair(request_id),
64 u32.clone.pair(operation), ByteMemTable.bytes.clone(payload))
65
66def status.clone.pair(
67 status: Protocol.ResponseStatus
68) -> Protocol.ResponseStatus & Protocol.ResponseStatus:
69 match status:
70 case Protocol.Success{}: (Protocol.Success{}, Protocol.Success{})
71 case Protocol.InvalidArgument{}:
72 (Protocol.InvalidArgument{}, Protocol.InvalidArgument{})
73 case Protocol.NotFound{}: (Protocol.NotFound{}, Protocol.NotFound{})
74 case Protocol.Busy{}: (Protocol.Busy{}, Protocol.Busy{})
75 case Protocol.Timeout{}: (Protocol.Timeout{}, Protocol.Timeout{})
76 case Protocol.ResponseResourceLimit{}:
77 (Protocol.ResponseResourceLimit{}, Protocol.ResponseResourceLimit{})
78 case Protocol.Corruption{}: (Protocol.Corruption{}, Protocol.Corruption{})
79 case Protocol.IoFailure{}: (Protocol.IoFailure{}, Protocol.IoFailure{})
80 case Protocol.CommitUnknown{}:
81 (Protocol.CommitUnknown{}, Protocol.CommitUnknown{})
82 case Protocol.AlreadyExists{}:
83 (Protocol.AlreadyExists{}, Protocol.AlreadyExists{})
84 case Protocol.UnsupportedPlatform{}:
85 (Protocol.UnsupportedPlatform{}, Protocol.UnsupportedPlatform{})
86 case Protocol.InvalidUtf8{}: (Protocol.InvalidUtf8{}, Protocol.InvalidUtf8{})
87 case Protocol.UnsupportedFormat{}:
88 (Protocol.UnsupportedFormat{}, Protocol.UnsupportedFormat{})
89
90def response.clone.payload(
91 versions: U32 & U32,
92 generations: U32 & U32,
93 request_ids: U32 & U32,
94 statuses: Protocol.ResponseStatus & Protocol.ResponseStatus,
95 payloads: Bytes.Bytes & Bytes.Bytes
96) -> Protocol.Response & Protocol.Response:
97 match versions:
98 case (version, version_copy):
99 match generations:
100 case (generation, generation_copy):
101 match request_ids:
102 case (request_id, request_id_copy):
103 match statuses:
104 case (status, status_copy):
105 match payloads:
106 case (first, second):
107 (Protocol.Response{version, generation, request_id,
108 status, first},
109 Protocol.Response{version_copy, generation_copy,
110 request_id_copy, status_copy, second})
111
112def response.clone.pair(
113 response: Protocol.Response
114) -> Protocol.Response & Protocol.Response:
115 match response:
116 case Protocol.Response{version, generation, request_id, status, payload}:
117 response.clone.payload(u32.clone.pair(version),
118 u32.clone.pair(generation), u32.clone.pair(request_id),
119 status.clone.pair(status),
120 ByteMemTable.bytes.clone(payload))
121
122def result.failure(error: Error) -> Result<&1, &1, Error, Protocol.Response>:
123 Fail{error}
124
125def result.response(
126 response: Protocol.Response
127) -> Result<&1, &1, Error, Protocol.Response>:
128 Done{response}
129
130def disconnect.error(mutation: Bool, error: Io.Error) -> Error:
131 match mutation:
132 case True{}: CommitUnknown{error}
133 case False{}: RetryRead{error}
134
135def response.validation.fields(
136 response_fields: U32 & U32 & U32,
137 request_fields: U32 & U32 & U32,
138 generations: U32 & U32 & U32
139) -> Policy.Decision:
140 match response_fields:
141 case (version, (response_generation, request_id)):
142 match request_fields:
143 case (request_version, (request_generation, expected_id)):
144 match generations:
145 case (response_generation_check,
146 (request_generation_check, generation)):
147 Policy.response.validate(
148 U32.to_nat(generation),
149 U32.is_eq(response_generation, response_generation_check) &&
150 U32.is_eq(request_generation, request_generation_check),
151 U32.is_eq(version, request_version) &&
152 U32.is_eq(request_id, expected_id))
153
154def response.validation.request(
155 response: Protocol.Response,
156 request: Protocol.Request,
157 generations: U32 & U32 & U32
158) -> Policy.Decision:
159 match response:
160 case Protocol.Response{version, response_generation, request_id, _, _}:
161 match request:
162 case Protocol.Request{request_version, request_generation,
163 expected_id, _, _}:
164 response.validation.fields((version,
165 (response_generation, request_id)),
166 (request_version, (request_generation, expected_id)), generations)
167
168def response.validation(
169 response: Protocol.Response,
170 request: Protocol.Request,
171 generation: U32
172) -> Policy.Decision:
173 response.validation.request(response, request,
174 Protocol.u32.duplicate.triple(generation))
175
176def response.checked.result(
177 response: Protocol.Response,
178 decision: Policy.Decision
179) -> Result<&1, &1, Error, Protocol.Response>:
180 match decision:
181 case Policy.ClientConfirmed{_}: result.response(response)
182 case Policy.ClientStaleOwner{}: result.failure(StaleOwner{})
183 case Policy.ClientProtocolFailure{}: result.failure(CorrelationMismatch{})
184 case _: result.failure(CorrelationMismatch{})
185
186def response.checked.request_copy(
187 retained: Protocol.Response,
188 checked: Protocol.Response,
189 generation: U32,
190 requests: Protocol.Request & Protocol.Request
191) -> Result<&1, &1, Error, Protocol.Response>:
192 match requests:
193 case (_, checked_request):
194 response.checked.result(retained,
195 response.validation(checked, checked_request, generation))
196
197def response.checked.copies(
198 responses: Protocol.Response & Protocol.Response,
199 requests: Protocol.Request & Protocol.Request,
200 generation: U32
201) -> Result<&1, &1, Error, Protocol.Response>:
202 match responses:
203 case (retained, checked):
204 response.checked.request_copy(retained, checked, generation, requests)
205
206def response.checked(
207 response: Protocol.Response,
208 request: Protocol.Request,
209 generation: U32
210) -> Result<&1, &1, Error, Protocol.Response>:
211 response.checked.copies(response.clone.pair(response),
212 request.clone.pair(request), generation)
213
214def response.read(
215 pair: File & Result<&1, &1, Io.Error, Protocol.Response>,
216 request: Protocol.Request,
217 generation: U32,
218 mutation: Bool
219) -> IO(Result<&1, &1, Error, Protocol.Response>):
220 match pair:
221 case (client, Done{response}):
222 do IO<Result<&1, &1, Error, Protocol.Response>>:
223 _closed : Unit <- File.close(client)
224 return response.checked(response, request, generation)
225 case (client, Fail{error}):
226 do IO<Result<&1, &1, Error, Protocol.Response>>:
227 _closed : Unit <- File.close(client)
228 return result.failure(disconnect.error(mutation, error))
229
230def response.request(
231 client: File,
232 request: Protocol.Request,
233 generation: U32,
234 mutation: Bool
235) -> IO(Result<&1, &1, Error, Protocol.Response>):
236 do IO<Result<&1, &1, Error, Protocol.Response>>:
237 response : File & Result<&1, &1, Io.Error, Protocol.Response> <-
238 Io.read_response(client)
239 response.read(response, request, generation, mutation)
240
241# Sends a frame result into response reading or mutation uncertainty.
242def written(
243 pair: File & Result<&1, &1, U32 & String, Unit>,
244 request: Protocol.Request,
245 generation: U32,
246 mutation: Bool
247) -> IO(Result<&1, &1, Error, Protocol.Response>):
248 match pair:
249 case (client, Done{Unit{}}):
250 response.request(client, request, generation, mutation)
251 case (client, Fail{(code, message)}):
252 do IO<Result<&1, &1, Error, Protocol.Response>>:
253 _closed : Unit <- File.close(client)
254 return result.failure(disconnect.error(mutation,
255 Io.host.error(code, message)))
256
257def connected.requests(
258 client: File,
259 generation: U32,
260 mutation: Bool,
261 requests: Protocol.Request & Protocol.Request
262) -> IO(Result<&1, &1, Error, Protocol.Response>):
263 match requests:
264 case (wire_request, retained_request):
265 do IO<Result<&1, &1, Error, Protocol.Response>>:
266 write_result : File & Result<&1, &1, U32 & String, Unit> <-
267 Io.write_all(client, Protocol.encode(wire_request))
268 written(write_result, retained_request, generation, mutation)
269
270# Writes one request after the owner endpoint accepts the connection.
271def connected(
272 pair: File & Result<&1, &1, UnixSocket.Error, Unit>,
273 request: Protocol.Request,
274 generation: U32,
275 mutation: Bool
276) -> IO(Result<&1, &1, Error, Protocol.Response>):
277 match pair:
278 case (client, Done{Unit{}}):
279 connected.requests(client, generation, mutation,
280 request.clone.pair(request))
281 case (client, Fail{error}):
282 do IO<Result<&1, &1, Error, Protocol.Response>>:
283 _closed : Unit <- File.close(client)
284 return result.failure(ConnectFailure{error})
285
286# Connects a new socket and classifies failures before dispatch.
287def created(
288 result: Result<&1, &1, UnixSocket.Error, File>,
289 endpoint: String,
290 request: Protocol.Request,
291 generation: U32,
292 mutation: Bool
293) -> IO(Result<&1, &1, Error, Protocol.Response>):
294 match result:
295 case Done{client}:
296 do IO<Result<&1, &1, Error, Protocol.Response>>:
297 connection : File & Result<&1, &1, UnixSocket.Error, Unit> <-
298 UnixSocket.connect(client, endpoint)
299 connected(connection, request, generation, mutation)
300 case Fail{error}:
301 IO.pure(Result<&1, &1, Error, Protocol.Response>,
302 result.failure(ConnectFailure{error}))
303
304def call.admitted(
305 endpoint: String,
306 request: Protocol.Request,
307 generation: U32,
308 mutation: Bool,
309 admitted: Bool
310) -> IO(Result<&1, &1, Error, Protocol.Response>):
311 match admitted:
312 case False{}:
313 IO.pure(Result<&1, &1, Error, Protocol.Response>,
314 result.failure(FrameTooLarge{}))
315 case True{}:
316 do IO<Result<&1, &1, Error, Protocol.Response>>:
317 client : Result<&1, &1, UnixSocket.Error, File> <-
318 UnixSocket.create()
319 created(client, endpoint, request, generation, mutation)
320
321def call.request.size(
322 endpoint: String,
323 fields: U32 & U32 & U32 & U32 & U32,
324 generation: U32,
325 measured_payload: Bytes.Bytes & U32
326) -> IO(Result<&1, &1, Error, Protocol.Response>):
327 match fields:
328 case (version, (request_generation, (request_id,
329 (policy_operation, wire_operation)))):
330 match measured_payload:
331 case (payload, length):
332 call.admitted(endpoint,
333 Protocol.Request{version, request_generation, request_id,
334 wire_operation, payload}, generation,
335 Policy.operation.is_mutation(policy_operation),
336 Policy.request.within_payload_limit(length))
337
338# Sends one bounded owner request and never retries it implicitly.
339def call(
340 endpoint: String,
341 request: Protocol.Request,
342 generation: U32
343) -> IO(Result<&1, &1, Error, Protocol.Response>):
344 match request:
345 case Protocol.Request{version, request_generation, request_id, operation,
346 payload}:
347 call.request.size(endpoint,
348 (version, (request_generation, (request_id,
349 u32.clone.pair(operation)))), generation, Bytes.length(payload))
350
351def string.clone.pair.go(
352 value: String,
353 left: String,
354 right: String
355) -> String & String:
356 match value:
357 case SNil{}: (String.reverse(left), String.reverse(right))
358 case SCon{+character, rest}:
359 string.clone.pair.go(rest, SCon{character, left}, SCon{character, right})
360
361def string.clone.pair(value: String) -> String & String:
362 string.clone.pair.go(value, SNil{}, SNil{})
363
364# Starts one client session with its first request identifier.
365def session.open(endpoint: String, generation: U32) -> Session:
366 Session{endpoint, generation, Policy.RequestAvailable{1}}
367
368def session.clone.ready(
369 endpoints: String & String,
370 generations: U32 & U32,
371 requests: Policy.RequestSequence & Policy.RequestSequence
372) -> Session & Session:
373 match endpoints:
374 case (first_endpoint, second_endpoint):
375 match generations:
376 case (first_generation, second_generation):
377 match requests:
378 case (first_requests, second_requests):
379 (Session{first_endpoint, first_generation, first_requests},
380 Session{second_endpoint, second_generation, second_requests})
381
382def session.clone(session: Session) -> Session & Session:
383 match session:
384 case Session{endpoint, +generation, requests}:
385 session.clone.ready(string.clone.pair(endpoint),
386 (generation, generation), SessionPolicy.requests.clone(requests))
387
388def session.call.ready(
389 endpoints: String & String,
390 generations: U32 & U32 & U32,
391 request_id: U32,
392 operation: U32,
393 next: Policy.RequestSequence,
394 payload: Bytes.Bytes
395) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
396 match endpoints:
397 case (call_endpoint, retained_endpoint):
398 match generations:
399 case (request_generation, (active_generation, retained_generation)):
400 do IO<Session & Result<&1, &1, Error, Protocol.Response>>:
401 response : Result<&1, &1, Error, Protocol.Response> <-
402 call(call_endpoint,
403 Protocol.Request{1, request_generation, request_id,
404 operation, payload}, active_generation)
405 return (Session{retained_endpoint, retained_generation, next},
406 response)
407
408def session.call.decision(
409 endpoint: String,
410 generation: U32,
411 operation: U32,
412 payload: Bytes.Bytes,
413 decision: Policy.RequestDecision
414) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
415 match decision:
416 case Policy.RequestAssigned{id, next}:
417 session.call.ready(string.clone.pair(endpoint),
418 Protocol.u32.duplicate.triple(generation), id, operation, next, payload)
419 case Policy.RequestIdsExhausted{}:
420 IO.pure(Session & Result<&1, &1, Error, Protocol.Response>,
421 (Session{endpoint, generation, Policy.RequestExhausted{}},
422 Fail{RequestIdsExhausted{}}))
423
424# Sends one operation with the next unique ID and retains the updated session.
425def session.call(
426 session: Session,
427 operation: U32,
428 payload: Bytes.Bytes
429) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
430 match session:
431 case Session{endpoint, generation, requests}:
432 session.call.decision(endpoint, generation, operation, payload,
433 Policy.request_id.take(requests))
434
435# Sends a byte get and retains the next request sequence.
436def session.get(
437 session: Session,
438 key: Bytes.Bytes
439) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
440 session.call(session, 1, key)
441
442# Sends a byte put and retains the next request sequence.
443def session.put(
444 session: Session,
445 key: Bytes.Bytes,
446 value: Bytes.Bytes
447) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
448 session.call(session, 2, Operation.encode.put(key, value))
449
450# Sends a byte delete and retains the next request sequence.
451def session.delete(
452 session: Session,
453 key: Bytes.Bytes
454) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
455 session.call(session, 3, key)
456
457# Sends a flush operation and retains the next request sequence.
458def session.flush(
459 session: Session
460) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
461 session.call(session, 4, Bytes.new(0))
462
463# Sends a compaction operation and retains the next request sequence.
464def session.compact(
465 session: Session
466) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
467 session.call(session, 5, Bytes.new(0))
468
469# Sends a stats operation and retains the next request sequence.
470def session.stats(
471 session: Session
472) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
473 session.call(session, 6, Bytes.new(0))
474
475def session.describe_owner(
476 session: Session
477) -> IO(Session & Result<&1, &1, Error, Protocol.Response>):
478 session.call(session, 16, Bytes.new(0))