1	import Base
2	import ./ByteWal.bend as ByteWal
3	import ./StorageBytes.bend as StorageBytes
4	import ./Wal.bend as Wal
5	import ./coordinator/QueuePolicy.bend as QueuePolicy
6	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7	
8	# Carries the owned input and accumulated exact batch size through the fold.
9	type ByteBatchMeasureState is Type:
10	  MeasureNeed{mutations: List<&1, ByteWal.Mutation>,
11	    reversed: List<&1, ByteWal.Mutation>, count: Nat, record_bytes: Nat}
12	
13	# Returns a measured list or rejects input exceeding the mutation-count bound.
14	type ByteBatchMeasurement is Type:
15	  BatchMeasured{mutations: List<&1, ByteWal.Mutation>, count: Nat,
16	    encoded_bytes: Nat}
17	  BatchTooMany{}
18	
19	# Returns the count-limit decision for a mutation batch.
20	type Validation is Data:
21	  ValidBatch{}
22	  EmptyBatch{}
23	  BatchCountExceeded{}
24	  BatchBytesExceeded{}
25	
26	# Returns the maximum number of mutations in a batch.
27	def max_batch_count() -> Nat:
28	  256n
29	
30	# Caps an encoded multi-mutation WAL frame at the normal block size.
31	def max_batch_bytes() -> Nat:
32	  U32.to_nat(StorageBytes.MAX_BLOCK_BYTES())
33	
34	# Retains the codec's one-record exception to the normal frame limit.
35	def max_single_batch_bytes() -> Nat:
36	  Nat.add(U32.to_nat(StorageBytes.MAX_RECORD_BYTES()), 36n)
37	
38	# Selects the encoded-frame ceiling for a single record or mutation batch.
39	def frame_byte_limit(
40	  count: Nat,
41	  batch_limit: Nat,
42	  single_limit: Nat
43	) -> Nat:
44	  Bool.pick(Nat, Nat.is_eq(count, 1n), single_limit, batch_limit)
45	
46	# Counts one encoded Put record's tag, length fields, key, and value.
47	def put_record_bytes(key_bytes: Nat, value_bytes: Nat) -> Nat:
48	  Nat.add(9n, Nat.add(key_bytes, value_bytes))
49	
50	# Counts one encoded Delete record's tag, length field, and key.
51	def delete_record_bytes(key_bytes: Nat) -> Nat:
52	  Nat.add(5n, key_bytes)
53	
54	# Decides whether one encoded frame is within its selected byte ceiling.
55	def validate_frame_bytes(
56	  count: Nat,
57	  encoded_bytes: Nat,
58	  batch_limit: Nat,
59	  single_limit: Nat
60	) -> Validation:
61	  Bool.pick(Validation,
62	    Nat.is_le(encoded_bytes, frame_byte_limit(count, batch_limit, single_limit)),
63	    ValidBatch{}, BatchBytesExceeded{})
64	
65	# Adds the fixed generation-four WAL frame header and checksum size.
66	def frame_bytes(record_bytes: Nat) -> Nat:
67	  Nat.add(36n, record_bytes)
68	
69	# Advances a bounded no-copy byte-batch size pass.
70	def byte_batch.measure.loop(
71	  fuel: Nat,
72	  state: ByteBatchMeasureState
73	) -> ByteBatchMeasurement:
74	  match fuel:
75	    case 0n:
76	      match state:
77	        case MeasureNeed{Nil{}, reversed, count, record_bytes}:
78	          BatchMeasured{List.reverse(&1, ByteWal.Mutation, reversed), count,
79	            frame_bytes(record_bytes)}
80	        case MeasureNeed{Con{_, _}, _, _, _}: BatchTooMany{}
81	    case 1n+rest:
82	      match state:
83	        case MeasureNeed{Nil{}, reversed, count, record_bytes}:
84	          BatchMeasured{List.reverse(&1, ByteWal.Mutation, reversed), count,
85	            frame_bytes(record_bytes)}
86	        case MeasureNeed{Con{ByteWal.Delete{key}, tail}, reversed, count,
87	            record_bytes}:
88	          match key:
89	            case Bytes.Bytes{+key_len, key_buffer}:
90	              byte_batch.measure.loop(rest, MeasureNeed{tail,
91	                Con{ByteWal.Delete{Bytes.Bytes{key_len, key_buffer}}, reversed},
92	                Nat.add(count, 1n),
93	                Nat.add(record_bytes,
94	                  delete_record_bytes(U32.to_nat(key_len)))})
95	        case MeasureNeed{Con{ByteWal.Put{key, value}, tail}, reversed, count,
96	            record_bytes}:
97	          match key:
98	            case Bytes.Bytes{+key_len, key_buffer}:
99	              match value:
100	                case Bytes.Bytes{+value_len, value_buffer}:
101	                  byte_batch.measure.loop(rest, MeasureNeed{tail,
102	                    Con{ByteWal.Put{Bytes.Bytes{key_len, key_buffer},
103	                      Bytes.Bytes{value_len, value_buffer}}, reversed},
104	                    Nat.add(count, 1n), Nat.add(record_bytes,
105	                      put_record_bytes(U32.to_nat(key_len),
106	                        U32.to_nat(value_len)))})
107	
108	# Measures a byte batch without copying buffers or building its WAL frame.
109	def byte_batch.measure(
110	  mutations: List<&1, ByteWal.Mutation>
111	) -> ByteBatchMeasurement:
112	  byte_batch.measure.loop(max_batch_count(),
113	    MeasureNeed{mutations, Nil{}, 0n, 0n})
114	
115	# Checks whether a batch count is within the allowed range.
116	def within_batch_count(count: Nat) -> Bool:
117	  Nat.is_le(count, max_batch_count())
118	
119	# Rejects empty and oversized batches before writing.
120	def validate_count(count: Nat) -> Validation:
121	  match count:
122	    case 0n:
123	      EmptyBatch{}
124	    case count:
125	      Bool.pick(Validation, within_batch_count(count), ValidBatch{}, BatchCountExceeded{})
126	
127	# Applies the byte limit only after the mutation count is accepted.
128	def validate_batch.counted(
129	  count: Nat,
130	  encoded_bytes: Nat,
131	  count_validation: Validation
132	) -> Validation:
133	  match count_validation:
134	    case ValidBatch{}:
135	      validate_frame_bytes(count, encoded_bytes, max_batch_bytes(),
136	        max_single_batch_bytes())
137	    case EmptyBatch{}: EmptyBatch{}
138	    case BatchCountExceeded{}: BatchCountExceeded{}
139	    case BatchBytesExceeded{}: BatchBytesExceeded{}
140	
141	# Rejects a batch before storage access if its count or frame bytes exceed limits.
142	def validate_batch.copies(
143	  counts: Nat & Nat,
144	  encoded_bytes: Nat
145	) -> Validation:
146	  match counts:
147	    case (for_frame_limit, for_count_validation):
148	      validate_batch.counted(for_frame_limit, encoded_bytes,
149	        validate_count(for_count_validation))
150	
151	# Rejects a batch before storage access if its count or frame bytes exceed limits.
152	def validate_batch(count: Nat, encoded_bytes: Nat) -> Validation:
153	  validate_batch.copies(QueuePolicy.nat.clone.pair(count), encoded_bytes)