1	# Bounded packed-byte transfers between files and TCP/TLS sockets.
2	import Base
3	import bend-kit-files@0.2.0.0/files.bend as Files
4	import bend-kit-wire@0.4.6.1/wire.bend as Wire
5	
6	type Error is Data:
7	  BadChunk{}
8	  Limit{}
9	  Read{code: U32, message: String}
10	  Write{code: U32, message: String}
11	
12	type Outcome is Data:
13	  Complete{count: U32}
14	  Failed{count: U32, error: Error}
15	
16	# Internal pure decisions, separated from the trusted effects.
17	type ReadAction is Data:
18	  End{}
19	  Over{}
20	  Forward{}
21	
22	def chunk.valid(+chunk: U32) -> Bool:
23	  U32.is_ge(chunk, 1) && U32.is_le(chunk, 1048576)
24	
25	# The one-byte probe is only requested after the entire cap was forwarded.
26	def request(+chunk: U32, +remaining: U32) -> U32:
27	  match remaining:
28	    case 0:
29	      1
30	    case _:
31	      U32.min(chunk, remaining)
32	
33	def read.action.flags(zero: Bool, probe: Bool) -> ReadAction:
34	  match zero:
35	    case True{}:
36	      End{}
37	    case False{}:
38	      match probe:
39	        case True{}:
40	          Over{}
41	        case False{}:
42	          Forward{}
43	
44	def read.action(len: U32, remaining: U32) -> ReadAction:
45	  read.action.flags(U32.is_zero(len), U32.is_zero(remaining))
46	
47	type Transfer<-S: Type, -D: Type> is Type:
48	  More{source: S, destination: D, count: U32}
49	  Stop{source: S, destination: D, outcome: Outcome}
50	
51	def file.read(source: File, max: U32, tls: Bool, ms: U32) ->
52	  IO(File & Result<&1, &1, U32 & String, U32 & Array<U32>>):
53	  Files.read.words(source, max)
54	
55	def file.write(destination: File, len: U32, words: Array<U32>, tls: Bool) ->
56	  IO(File & Result<&1, &1, U32 & String, Unit>):
57	  Files.write.words(destination, len, words)
58	
59	def socket.read.tls(tls: Bool, source: Socket, max: U32, ms: U32) ->
60	  IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
61	  match tls:
62	    case True{}:
63	      Wire.tls.recv.words(source, max, ms)
64	    case False{}:
65	      Wire.recv.words(source, max, ms)
66	
67	def socket.read(source: Socket, max: U32, tls: Bool, ms: U32) ->
68	  IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
69	  socket.read.tls(tls, source, max, ms)
70	
71	def socket.write.tls(tls: Bool, destination: Socket, len: U32, words: Array<U32>) ->
72	  IO(Socket & Result<&1, &1, U32 & String, Unit>):
73	  match tls:
74	    case True{}:
75	      Wire.tls.send.words(destination, len, words)
76	    case False{}:
77	      Wire.send.words(destination, len, words)
78	
79	def socket.write(destination: Socket, len: U32, words: Array<U32>, tls: Bool) ->
80	  IO(Socket & Result<&1, &1, U32 & String, Unit>):
81	  socket.write.tls(tls, destination, len, words)
82	
83	def transfer.written(~S: Type, ~D: Type, source: S, count: U32, len: U32,
84	  pair: D & Result<&1, &1, U32 & String, Unit>) -> Transfer<S, D>:
85	  (destination, result) = pair
86	  match result:
87	    case Fail{(code, message)}:
88	      Stop{source, destination, Failed{count, Write{code, message}}}
89	    case Done{unit}:
90	      More{source, destination, (count + len : U32)}
91	
92	def transfer.action(~S: Type, ~D: Type,
93	  ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
94	  action: ReadAction, source: S, destination: D, count: U32, +len: U32,
95	  words: Array<U32>, tls: Bool) -> IO(Transfer<S, D>):
96	  match action:
97	    case End{}:
98	      IO.pure(Transfer<S, D>, Stop{source, destination, Complete{count}})
99	    case Over{}:
100	      IO.pure(Transfer<S, D>, Stop{source, destination, Failed{count, Limit{}}})
101	    case Forward{}:
102	      do IO<Transfer<S, D>>:
103	        pair : D & Result<&1, &1, U32 & String, Unit> <- write(destination, len, words, tls)
104	        return transfer.written(~S, ~D, source, count, len, pair)
105	
106	def transfer.read(~S: Type, ~D: Type,
107	  ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
108	  pair: S & Result<&1, &1, U32 & String, U32 & Array<U32>>,
109	  destination: D, count: U32, remaining: U32, tls: Bool) -> IO(Transfer<S, D>):
110	  (source, result) = pair
111	  match result:
112	    case Fail{(code, message)}:
113	      IO.pure(Transfer<S, D>, Stop{source, destination, Failed{count, Read{code, message}}})
114	    case Done{(+len, words)}:
115	      transfer.action(~S, ~D, ~write, read.action(len, remaining),
116	        source, destination, count, len, words, tls)
117	
118	# cap+1 reads suffice even if every successful nonempty read contains one byte.
119	# Fuel is first among runtime parameters; no fixed read ceiling or unsafe recursion.
120	def transfer.loop(~S: Type, ~D: Type,
121	  ~read: S -> U32 -> Bool -> U32 -> IO(S & Result<&1, &1, U32 & String, U32 & Array<U32>>),
122	  ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
123	  fuel: Nat, state: Transfer<S, D>, +tls: Bool, +chunk: U32, +cap: U32, +ms: U32) ->
124	  IO(S & D & Outcome):
125	  match fuel:
126	    case 0n:
127	      match state:
128	        case Stop{source, destination, outcome}:
129	          IO.pure(S & D & Outcome, (source, destination, outcome))
130	        case More{source, destination, count}:
131	          # Unreachable under the read.words bounds and positive-progress contract.
132	          IO.pure(S & D & Outcome, (source, destination, Failed{count, Limit{}}))
133	    case 1n+rest:
134	      match state:
135	        case Stop{source, destination, outcome}:
136	          IO.pure(S & D & Outcome, (source, destination, outcome))
137	        case More{source, destination, +count}:
138	          +remaining = (cap - count : U32)
139	          do IO<S & D & Outcome>:
140	            pair : S & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
141	              read(source, request(chunk, remaining), tls, ms)
142	            next : Transfer<S, D> <- transfer.read(~S, ~D, ~write,
143	              pair, destination, count, remaining, tls)
144	            transfer.loop(~S, ~D, ~read, ~write, rest, next, tls, chunk, cap, ms)
145	
146	def transfer.start(~S: Type, ~D: Type,
147	  ~read: S -> U32 -> Bool -> U32 -> IO(S & Result<&1, &1, U32 & String, U32 & Array<U32>>),
148	  ~write: D -> U32 -> Array<U32> -> Bool -> IO(D & Result<&1, &1, U32 & String, Unit>),
149	  valid: Bool, source: S, destination: D, tls: Bool, chunk: U32, +cap: U32, ms: U32) ->
150	  IO(S & D & Outcome):
151	  match valid:
152	    case False{}:
153	      IO.pure(S & D & Outcome, (source, destination, Failed{0, BadChunk{}}))
154	    case True{}:
155	      transfer.loop(~S, ~D, ~read, ~write, 1n+U32.to_nat(cap),
156	        More{source, destination, 0}, tls, chunk, cap, ms)
157	
158	# Return both handles on every outcome; never close or retry a failed chunk.
159	def file.file(source: File, destination: File, +chunk: U32, cap: U32) ->
160	  IO(File & File & Outcome):
161	  transfer.start(~File, ~File, ~file.read, ~file.write,
162	    chunk.valid(chunk), source, destination, False{}, chunk, cap, 0)
163	
164	def file.socket(source: File, destination: Socket, tls: Bool, +chunk: U32, cap: U32) ->
165	  IO(File & Socket & Outcome):
166	  transfer.start(~File, ~Socket, ~file.read, ~socket.write,
167	    chunk.valid(chunk), source, destination, tls, chunk, cap, 0)
168	
169	# ms is the unchanged per-read Wire deadline, including the one-byte cap probe.
170	def socket.file(source: Socket, destination: File, tls: Bool, +chunk: U32, cap: U32, ms: U32) ->
171	  IO(Socket & File & Outcome):
172	  transfer.start(~Socket, ~File, ~socket.read, ~file.write,
173	    chunk.valid(chunk), source, destination, tls, chunk, cap, ms)