1 import Base 2 import ../ByteMemTable.bend as ByteMemTable 3 import ./ClientPolicy.bend as Policy 4 import ./SessionPolicy.bend as SessionPolicy 5 import ./Io.bend as Io 6 import ./Operation.bend as Operation 7 import ./Protocol.bend as Protocol 8 import ./UnixSocketIo.bend as UnixSocket 9 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 10 11 # Returns connection, transport, correlation, and retry-boundary outcomes. 12 type 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. 24 type Session is Type: 25 Session{endpoint: String, generation: U32, 26 requests: Policy.RequestSequence} 27 28 def u32.clone.pair.from(triple: U32 & U32 & U32) -> U32 & U32: 29 match triple: 30 case (first, (second, _)): (first, second) 31 32 def u32.clone.pair(value: U32) -> U32 & U32: 33 u32.clone.pair.from(Protocol.u32.duplicate.triple(value)) 34 35 def 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 57 def 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 66 def 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 90 def 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 112 def 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 122 def result.failure(error: Error) -> Result<&1, &1, Error, Protocol.Response>: 123 Fail{error} 124 125 def result.response( 126 response: Protocol.Response 127 ) -> Result<&1, &1, Error, Protocol.Response>: 128 Done{response} 129 130 def disconnect.error(mutation: Bool, error: Io.Error) -> Error: 131 match mutation: 132 case True{}: CommitUnknown{error} 133 case False{}: RetryRead{error} 134 135 def 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 154 def 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 168 def 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 176 def 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 186 def 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 197 def 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 206 def 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 214 def 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>: 223 _closed : Unit <- File.close(client) 224 return response.checked(response, request, generation) 225 case (client, Fail{error}): 226 do IO>: 227 _closed : Unit <- File.close(client) 228 return result.failure(disconnect.error(mutation, error)) 229 230 def 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>: 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. 242 def 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>: 253 _closed : Unit <- File.close(client) 254 return result.failure(disconnect.error(mutation, 255 Io.host.error(code, message))) 256 257 def 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>: 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. 271 def 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>: 283 _closed : Unit <- File.close(client) 284 return result.failure(ConnectFailure{error}) 285 286 # Connects a new socket and classifies failures before dispatch. 287 def 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>: 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 304 def 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>: 317 client : Result<&1, &1, UnixSocket.Error, File> <- 318 UnixSocket.create() 319 created(client, endpoint, request, generation, mutation) 320 321 def 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. 339 def 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 351 def 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 361 def 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. 365 def session.open(endpoint: String, generation: U32) -> Session: 366 Session{endpoint, generation, Policy.RequestAvailable{1}} 367 368 def 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 382 def 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 388 def 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>: 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 408 def 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. 425 def 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. 436 def 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. 443 def 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. 451 def 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. 458 def 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. 464 def 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. 470 def session.stats( 471 session: Session 472 ) -> IO(Session & Result<&1, &1, Error, Protocol.Response>): 473 session.call(session, 6, Bytes.new(0)) 474 475 def session.describe_owner( 476 session: Session 477 ) -> IO(Session & Result<&1, &1, Error, Protocol.Response>): 478 session.call(session, 16, Bytes.new(0))