1	# Bounded affine dependency bundles with typed concurrent operations and explicit close.
2	import Base
3	
4	type Counts is Data:
5	  Counts{capacity: Nat, in_use: Nat, completed: Nat, rejected: Nat}
6	
7	type Outcome<-E: Data, -O: Data> is Data:
8	  Success{value: O}
9	  Expected{error: E}
10	  Exhausted{}
11	  Stopped{}
12	
13	type CloseResult is Data:
14	  Closed{}
15	  Busy{counts: Counts}
16	
17	type Message<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Type:
18	  Submit{principal: P, input: I, reply: Chan(Outcome<E, O>)}
19	  Returned{resource: R}
20	  Inspect{reply: Chan(Counts)}
21	  Shutdown{reply: Chan(CloseResult)}
22	
23	# Copy only this context, never the resources. The inbox bounds queued commands;
24	# callers must count blocked submissions inside their transport admission limit.
25	type Context<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Data:
26	  Context{messages: Chan(Message<R, P, I, E, O>)}
27	
28	def outcome.value(-E: Data, -O: Data, result: Result<&2, &2, E, O>) -> Outcome<E, O>:
29	  match result:
30	    case Fail{error}:
31	      Expected{error}
32	    case Done{value}:
33	      Success{value}
34	
35	# The resources return through the private owner protocol BEFORE a disposable
36	# caller reply. A closed caller reply cannot erase a handle or strand capacity.
37	def work.return(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
38	  messages: Chan(Message<R, P, I, E, O>), reply: Chan(Outcome<E, O>),
39	  result: R & Result<&2, &2, E, O>
40	) -> IO(Unit):
41	  (returned, outcome) = result
42	  do IO<Unit>:
43	    accepted : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Returned{returned})
44	    sent : Bool <- Chan.send(Outcome<E, O>, reply, outcome.value(E, O, outcome))
45	    return Unit{}
46	
47	def work(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
48	  ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>),
49	  resource: R, principal: P, input: I,
50	  messages: Chan(Message<R, P, I, E, O>), reply: Chan(Outcome<E, O>)
51	) -> IO(Unit):
52	  do IO<Unit>:
53	    result : R & Result<&2, &2, E, O> <- handler(resource, principal, input)
54	    work.return(R, P, I, E, O, messages, reply, result)
55	
56	def dispose(~R: Type, ~destroy: R -> IO(Unit), resources: List<R>) -> IO(Unit):
57	  match resources:
58	    case Nil{}:
59	      IO.pure(Unit, Unit{})
60	    case Con{resource, rest}:
61	      do IO<Unit>:
62	        destroy(resource)
63	        dispose(~R, ~destroy, rest)
64	
65	# Closing Base channels retains buffered messages. Drain them and answer stopped
66	# callers, rather than stranding accepted commands behind the close command.
67	def drain.message(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
68	  ~destroy: R -> IO(Unit), message: Message<R, P, I, E, O>
69	) -> IO(Unit):
70	  match message:
71	    case Submit{principal, input, reply}:
72	      do IO<Unit>:
73	        sent : Bool <- Chan.send(Outcome<E, O>, reply, Stopped{})
74	        return Unit{}
75	    case Inspect{reply}:
76	      Chan.close(Counts, reply)
77	    case Shutdown{reply}:
78	      do IO<Unit>:
79	        sent : Bool <- Chan.send(CloseResult, reply, Closed{})
80	        return Unit{}
81	    case Returned{resource}:
82	      destroy(resource)
83	
84	def drain.step(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
85	  ~destroy: R -> IO(Unit), pending: Maybe<&1, Message<R, P, I, E, O>>,
86	  next: Unit -> IO(Unit)
87	) -> IO(Unit):
88	  match pending:
89	    case None{}:
90	      IO.pure(Unit, Unit{})
91	    case Some{message}:
92	      do IO<Unit>:
93	        drain.message(~R, ~P, ~I, ~E, ~O, ~destroy, message)
94	        next(Unit{})
95	
96	# The closed inbox contains at most room buffered messages.
97	def drain(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
98	  ~destroy: R -> IO(Unit), fuel: Nat, +messages: Chan(Message<R, P, I, E, O>)
99	) -> IO(Unit):
100	  match fuel:
101	    case 0n:
102	      IO.pure(Unit, Unit{})
103	    case 1n+p:
104	      do IO<Unit>:
105	        pending : Maybe<&1, Message<R, P, I, E, O>> <- Chan.recv(Message<R, P, I, E, O>, messages)
106	        drain.step(~R, ~P, ~I, ~E, ~O, ~destroy, pending, u => drain(~R, ~P, ~I, ~E, ~O, ~destroy, p, messages))
107	
108	type OwnerStep<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Type:
109	  Waiting{resources: List<R>, counts: Counts}
110	  Received{message: Message<R, P, I, E, O>, counts: Counts, resources: List<R>}
111	
112	# IO.join closes a one-shot channel; the owner inbox must remain reusable.
113	def receive.value(-A: Type, value: Maybe<&1, A>) -> IO(A):
114	  match value:
115	    case None{}:
116	      IO.die(A, 1, "camber internal channel closed unexpectedly")
117	    case Some{message}:
118	      IO.pure(A, message)
119	
120	def receive(-A: Type, channel: Chan(A)) -> IO(A):
121	  do IO<A>:
122	    value : Maybe<&1, A> <- Chan.recv(A, channel)
123	    receive.value(A, value)
124	
125	# Only idle close ends this loop. Busy close never waits on a stalled handler.
126	# @unsafe is solely the lifecycle receive loop, not a termination/proof claim.
127	@unsafe def owner(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
128	  ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
129	  +messages: Chan(Message<R, P, I, E, O>), +room: U32, step: OwnerStep<R, P, I, E, O>
130	) -> IO(Unit):
131	  match step:
132	    case Waiting{resources, counts}:
133	      do IO<Unit>:
134	        message : Message<R, P, I, E, O> <- receive(Message<R, P, I, E, O>, messages)
135	        owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Received{message, counts, resources})
136	    case Received{message, counts, resources}:
137	      match message:
138	        case Returned{resource}:
139	          Counts{capacity, in_use, completed, rejected} = counts
140	          owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resource <> resources, Counts{capacity, Nat.sub(in_use, 1n), 1n+completed, rejected}})
141	        case Inspect{reply}:
142	          Counts{+capacity, +in_use, +completed, +rejected} = counts
143	          do IO<Unit>:
144	            sent : Bool <- Chan.send(Counts, reply, Counts{capacity, in_use, completed, rejected})
145	            owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, in_use, completed, rejected}})
146	        case Shutdown{reply}:
147	          Counts{+capacity, in_use, +completed, +rejected} = counts
148	          match in_use:
149	            case 0n:
150	              do IO<Unit>:
151	                dispose(~R, ~destroy, resources)
152	                Chan.close(Message<R, P, I, E, O>, messages)
153	                sent : Bool <- Chan.send(CloseResult, reply, Closed{})
154	                drain(~R, ~P, ~I, ~E, ~O, ~destroy, U32.to_nat(room), messages)
155	            case 1n+ +p:
156	              do IO<Unit>:
157	                sent : Bool <- Chan.send(CloseResult, reply, Busy{Counts{capacity, 1n+p, completed, rejected}})
158	                owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, 1n+p, completed, rejected}})
159	        case Submit{principal, input, reply}:
160	          Counts{capacity, in_use, completed, rejected} = counts
161	          match resources:
162	            case Nil{}:
163	              do IO<Unit>:
164	                sent : Bool <- Chan.send(Outcome<E, O>, reply, Exhausted{})
165	                owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{Nil{}, Counts{capacity, in_use, completed, 1n+rejected}})
166	            case Con{resource, rest}:
167	              do IO<Unit>:
168	                IO.spawn(Unit, work(~R, ~P, ~I, ~E, ~O, ~handler, resource, principal, input, messages, reply))
169	                owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{rest, Counts{capacity, 1n+in_use, completed, rejected}})
170	
171	def build.step(~R: Type, ~E: Data, ~destroy: R -> IO(Unit),
172	  result: Result<&2, &1, E, R>, resources: List<R>,
173	  next: List<R> -> IO(Result<&2, &1, E, List<R>>)
174	) -> IO(Result<&2, &1, E, List<R>>):
175	  match result:
176	    case Fail{error}:
177	      do IO<Result<&2, &1, E, List<R>>>:
178	        dispose(~R, ~destroy, resources)
179	        return Fail{error}
180	    case Done{resource}:
181	      next(resource <> resources)
182	
183	def build(~K: Data, ~R: Type, ~E: Data, ~create: K -> Nat -> IO(Result<&2, &1, E, R>), ~destroy: R -> IO(Unit),
184	  remaining: Nat, +config: K, resources: List<R>
185	) -> IO(Result<&2, &1, E, List<R>>):
186	  match remaining:
187	    case 0n:
188	      IO.pure(Result<&2, &1, E, List<R>>, Done{resources})
189	    case 1n+ +p:
190	      do IO<Result<&2, &1, E, List<R>>>:
191	        result : Result<&2, &1, E, R> <- create(config, p)
192	        build.step(~R, ~E, ~destroy, result, resources, next => build(~K, ~R, ~E, ~create, ~destroy, p, config, next))
193	
194	def start.ready(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
195	  ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
196	  capacity: Nat, +room: U32, result: Result<&2, &1, E, List<R>>
197	) -> IO(Result<&2, &2, E, Context<R, P, I, E, O>>):
198	  match result:
199	    case Fail{error}:
200	      IO.pure(Result<&2, &2, E, Context<R, P, I, E, O>>, Fail{error})
201	    case Done{resources}:
202	      do IO<Result<&2, &2, E, Context<R, P, I, E, O>>>:
203	        channel : Chan(Message<R, P, I, E, O>) <- Chan.new(Message<R, P, I, E, O>, room)
204	        +messages : Chan(Message<R, P, I, E, O>) = channel
205	        IO.spawn(Unit, owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, 0n, 0n, 0n}}))
206	        return Done{Context{messages}}
207	
208	# The initializer closes partial handles within a failed bundle; start closes
209	# all prior whole bundles. No workers exist until the complete set opens.
210	def start(~K: Data, ~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
211	  ~create: K -> Nat -> IO(Result<&2, &1, E, R>),
212	  ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
213	  config: K, +capacity: Nat, room: U32
214	) -> IO(Result<&2, &2, E, Context<R, P, I, E, O>>):
215	  do IO<Result<&2, &2, E, Context<R, P, I, E, O>>>:
216	    result : Result<&2, &1, E, List<R>> <- build(~K, ~R, ~E, ~create, ~destroy, capacity, config, Nil{})
217	    start.ready(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, capacity, room, result)
218	
219	def submit(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
220	  context: Context<R, P, I, E, O>, principal: P, input: I, reply: Chan(Outcome<E, O>)
221	) -> IO(Bool):
222	  Context{messages} = context
223	  Chan.send(Message<R, P, I, E, O>, messages, Submit{principal, input, reply})
224	
225	def call.accepted(-E: Data, -O: Data, accepted: Bool, +reply: Chan(Outcome<E, O>)) -> IO(Outcome<E, O>):
226	  match accepted:
227	    case False{}:
228	      do IO<Outcome<E, O>>:
229	        Chan.close(Outcome<E, O>, reply)
230	        return Stopped{}
231	    case True{}:
232	      do IO<Outcome<E, O>>:
233	        result : Outcome<E, O> <- IO.join(Outcome<E, O>, reply)
234	        Chan.close(Outcome<E, O>, reply)
235	        return result
236	
237	def call(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
238	  context: Context<R, P, I, E, O>, principal: P, input: I
239	) -> IO(Outcome<E, O>):
240	  do IO<Outcome<E, O>>:
241	    channel : Chan(Outcome<E, O>) <- Chan.new(Outcome<E, O>, 1)
242	    +reply : Chan(Outcome<E, O>) = channel
243	    accepted : Bool <- submit(R, P, I, E, O, context, principal, input, reply)
244	    call.accepted(E, O, accepted, reply)
245	
246	def inspect.value(value: Maybe<&1, Counts>) -> Maybe<&2, Counts>:
247	  match value:
248	    case None{}:
249	      None{}
250	    case Some{counts}:
251	      Some{counts}
252	
253	def inspect.sent(sent: Bool, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
254	  match sent:
255	    case False{}:
256	      do IO<Maybe<&2, Counts>>:
257	        Chan.close(Counts, reply)
258	        return None{}
259	    case True{}:
260	      do IO<Maybe<&2, Counts>>:
261	        value : Maybe<&1, Counts> <- Chan.recv(Counts, reply)
262	        Chan.close(Counts, reply)
263	        return inspect.value(value)
264	
265	def inspect(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data, context: Context<R, P, I, E, O>) -> IO(Maybe<&2, Counts>):
266	  Context{messages} = context
267	  do IO<Maybe<&2, Counts>>:
268	    channel : Chan(Counts) <- Chan.new(Counts, 1)
269	    +reply : Chan(Counts) = channel
270	    sent : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Inspect{reply})
271	    inspect.sent(sent, reply)
272	
273	def close.sent(sent: Bool, +reply: Chan(CloseResult)) -> IO(CloseResult):
274	  match sent:
275	    case False{}:
276	      do IO<CloseResult>:
277	        Chan.close(CloseResult, reply)
278	        return Closed{}
279	    case True{}:
280	      do IO<CloseResult>:
281	        value : CloseResult <- IO.join(CloseResult, reply)
282	        Chan.close(CloseResult, reply)
283	        return value
284	
285	# Stop external admissions before close. Busy preserves live context/capacity;
286	# retry after admitted operations finish. Stalled operations remain observable.
287	def close(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data, context: Context<R, P, I, E, O>) -> IO(CloseResult):
288	  Context{messages} = context
289	  do IO<CloseResult>:
290	    channel : Chan(CloseResult) <- Chan.new(CloseResult, 1)
291	    +reply : Chan(CloseResult) = channel
292	    sent : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Shutdown{reply})
293	    close.sent(sent, reply)