1	import Base
2	import ./QueuePolicy.bend as QueuePolicy
3	import ./Protocol.bend as Protocol
4	
5	# Represents whether one request is ready, dispatched, complete, or stale.
6	type 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.
12	type 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.
23	type RequestSequence is Data:
24	  RequestAvailable{id: U32}
25	  RequestExhausted{}
26	
27	# Assigns one identifier and returns the next sequence state.
28	type RequestDecision is Type:
29	  RequestAssigned{id: U32, next: RequestSequence}
30	  RequestIdsExhausted{}
31	
32	# Classifies owner read statuses as a value, absence, or typed failure.
33	type GetStatus is Data:
34	  GetValueStatus{}
35	  GetMissingStatus{}
36	  GetFailureStatus{status: Protocol.ResponseStatus}
37	
38	def 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	
46	def 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	
55	def response.generation(
56	  generation: Nat,
57	  current_generation: Nat
58	) -> Decision:
59	  response.generation.checked(QueuePolicy.nat.clone.pair(generation),
60	    current_generation)
61	
62	def 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	
74	def 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.
80	def 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.
90	def 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.
94	def response.accepted(status: Protocol.ResponseStatus) -> Bool:
95	  match status:
96	    case Protocol.Success{}: True{}
97	    case _: False{}
98	
99	def 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	
105	def 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	
115	def 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.
124	def 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.
131	def 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.
138	def 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.
145	def disconnect(state: State) -> Decision:
146	  match state:
147	    case ClientReady{_}: ClientRetryBeforeDispatch{}
148	    case RequestDispatched{_, mutation}:
149	      disconnect.dispatched(mutation)
150	    case _: ClientReject{}