src/coordinator/ClientPolicy.bend

mylsm-lsm-store@0.6.0.1 · 150 lines · 20 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
3import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol
4
5# Represents whether one request is ready, dispatched, complete, or stale.
6type State is Data:
7  ClientReady{generation: Nat}
8  RequestDispatched{generation: Nat, mutation: Bool}
9  RequestCompleted{generation: Nat}
10
11# Separates confirmation, safe retry, stale-owner, and uncertain-write outcomes.
12type Decision is Data:
13  ClientDispatch{state: State}
14  ClientConfirmed{state: State}
15  ClientRetryRead{}
16  ClientRetryBeforeDispatch{}
17  ClientCommitUnknown{}
18  ClientStaleOwner{}
19  ClientProtocolFailure{}
20  ClientReject{}
21
22# Tracks request identifiers without allowing U32 wraparound reuse.
23type RequestSequence is Data:
24  RequestAvailable{id: U32}
25  RequestExhausted{}
26
27# Assigns one identifier and returns the next sequence state.
28type RequestDecision is Type:
29  RequestAssigned{id: U32, next: RequestSequence}
30  RequestIdsExhausted{}
31
32# Classifies owner read statuses as a value, absence, or typed failure.
33type GetStatus is Data:
34  GetValueStatus{}
35  GetMissingStatus{}
36  GetFailureStatus{status: Protocol.ResponseStatus}
37
38def response.generation.result(
39  generation: Nat,
40  generation_matches: Bool
41) -> Decision:
42  match generation_matches:
43    case True{}: ClientConfirmed{RequestCompleted{generation}}
44    case False{}: ClientStaleOwner{}
45
46def response.generation.checked(
47  generations: Nat & Nat,
48  current_generation: Nat
49) -> Decision:
50  match generations:
51    case (generation, checked_generation):
52      response.generation.result(generation,
53        Nat.is_eq(checked_generation, current_generation))
54
55def response.generation(
56  generation: Nat,
57  current_generation: Nat
58) -> Decision:
59  response.generation.checked(QueuePolicy.nat.clone.pair(generation),
60    current_generation)
61
62def response.validate(
63  generation: Nat,
64  generation_matches: Bool,
65  correlation_matches: Bool
66) -> Decision:
67  match generation_matches:
68    case False{}: ClientStaleOwner{}
69    case True{}:
70      match correlation_matches:
71        case True{}: ClientConfirmed{RequestCompleted{generation}}
72        case False{}: ClientProtocolFailure{}
73
74def disconnect.dispatched(mutation: Bool) -> Decision:
75  match mutation:
76    case True{}: ClientCommitUnknown{}
77    case False{}: ClientRetryRead{}
78
79# Identifies request tags that can change durable database state.
80def operation.is_mutation(operation: U32) -> Bool:
81  match operation:
82    case 2: True{}
83    case 3: True{}
84    case 4: True{}
85    case 5: True{}
86    case 8: True{}
87    case _: False{}
88
89# Enforces the encoded IPC payload ceiling before client socket allocation.
90def request.within_payload_limit(payload_bytes: U32) -> Bool:
91  Protocol.payload.admitted(payload_bytes)
92
93# Accepts only owner responses that confirm successful operation completion.
94def response.accepted(status: Protocol.ResponseStatus) -> Bool:
95  match status:
96    case Protocol.Success{}: True{}
97    case _: False{}
98
99def response.get.status(status: Protocol.ResponseStatus) -> GetStatus:
100  match status:
101    case Protocol.Success{}: GetValueStatus{}
102    case Protocol.NotFound{}: GetMissingStatus{}
103    case _: GetFailureStatus{status}
104
105def request_id.take.available(
106  ids: U32 & U32,
107  is_final: Bool
108) -> RequestDecision:
109  match ids:
110    case (id, next_id):
111      match is_final:
112        case True{}: RequestAssigned{id, RequestExhausted{}}
113        case False{}: RequestAssigned{id, RequestAvailable{(next_id + 1 : U32)}}
114
115def request_id.take.ready(
116  copies: U32 & U32 & U32
117) -> RequestDecision:
118  match copies:
119    case (final_check, (id, next_id)):
120      request_id.take.available((id, next_id),
121        U32.is_eq(final_check, 4294967295))
122
123# Assigns each request ID once and makes exhaustion terminal.
124def request_id.take(sequence: RequestSequence) -> RequestDecision:
125  match sequence:
126    case RequestAvailable{id}:
127      request_id.take.ready(Protocol.u32.duplicate.triple(id))
128    case RequestExhausted{}: RequestIdsExhausted{}
129
130# Marks a ready request as dispatched exactly once.
131def dispatch(state: State, mutation: Bool) -> Decision:
132  match state:
133    case ClientReady{generation}:
134      ClientDispatch{RequestDispatched{generation, mutation}}
135    case _: ClientReject{}
136
137# Resolves a response only when it belongs to the current owner generation.
138def response(state: State, current_generation: Nat) -> Decision:
139  match state:
140    case RequestDispatched{generation, _}:
141      response.generation(generation, current_generation)
142    case _: ClientReject{}
143
144# Classifies a disconnected request without retrying a dispatched mutation.
145def disconnect(state: State) -> Decision:
146  match state:
147    case ClientReady{_}: ClientRetryBeforeDispatch{}
148    case RequestDispatched{_, mutation}:
149      disconnect.dispatched(mutation)
150    case _: ClientReject{}