src/coordinator/ClientPolicy.bend
mylsm-lsm-store@0.6.0.0 · 150 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as QueuePolicy
3import mylsm-lsm-store@0.6.0.0/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{}