1 import Base 2 import ../ByteMemTable.bend as ByteMemTable 3 import ../scan/VisibleCursor.bend as VisibleCursor 4 import ./Backup.bend as Backup 5 import ./BackupCursorPolicy.bend as Policy 6 import ../coordinator/CursorCodec.bend as CursorCodec 7 import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes 8 9 # Separates bounded record emission from footer emission and completion. 10 type Phase is Data: 11 CursorBlocks{} 12 CursorFooter{} 13 CursorFinished{} 14 15 # Tracks one visible cursor and no more than one pending export block. 16 type Stream is Type: 17 Stream{phase: Phase, cursor: Maybe<&1, VisibleCursor.Cursor>, 18 pending: List<&1, ByteMemTable.Entry>, pending_count: Nat, 19 seen: Nat, expected: Nat, entry_count: U32, block_count: U32, 20 block_index: U32} 21 22 # Returns a bounded backup chunk, terminal completion, or a validation error. 23 type Step is Type: 24 Chunk{stream: Stream, bytes: Bytes.Bytes} 25 Continue{stream: Stream} 26 End{} 27 Failure{cursor: Maybe<&1, VisibleCursor.Cursor>, error: Backup.Error} 28 29 # Initializes the bounded cursor stream after encoding its header. 30 def start.ready( 31 cursor: Maybe<&1, VisibleCursor.Cursor>, 32 entry_count: U32, 33 expected: Nat, 34 block_count: U32, 35 header: Bytes.Bytes 36 ) -> Step: 37 Chunk{Stream{CursorBlocks{}, cursor, Nil{}, 0n, 0n, expected, 38 entry_count, block_count, 0}, header} 39 40 def start.count( 41 cursor: Maybe<&1, VisibleCursor.Cursor>, 42 entries: U32 & U32, 43 block_count: U32, 44 header: Bytes.Bytes 45 ) -> Step: 46 match entries: 47 case (entry_count, expected): 48 start.ready(cursor, entry_count, U32.to_nat(expected), 49 block_count, header) 50 51 def start.header( 52 cursor: Maybe<&1, VisibleCursor.Cursor>, 53 pair: Backup.Stream & Bytes.Bytes 54 ) -> Step: 55 match pair: 56 case (Backup.Stream{_, _, entry_count, block_count, _}, header): 57 start.count(cursor, Backup.u32.duplicate.pair(entry_count), 58 block_count, header) 59 60 def start.counts( 61 cursor: Maybe<&1, VisibleCursor.Cursor>, 62 counts: Nat & Nat 63 ) -> Step: 64 match counts: 65 case (entry_count, block_input): 66 +encoded_entries = U32.from_nat(entry_count) 67 +encoded_blocks = Backup.blocks.count(block_input) 68 start.header(cursor, Backup.header.bytes( 69 Backup.Stream{Backup.Header{}, Nil{}, encoded_entries, 70 encoded_blocks, 0})) 71 72 # Emits the format header and prepares the first incremental source read. 73 def start(cursor: VisibleCursor.Cursor, entry_count: Nat) -> Step: 74 start.counts(Some{cursor}, Backup.nat.duplicate.pair(entry_count)) 75 76 # Initializes the same bounded encoder without a local cursor handle. 77 def start.owner(entry_count: Nat) -> Step: 78 start.counts(None{}, Backup.nat.duplicate.pair(entry_count)) 79 80 def block.written( 81 result: Result<&1, &1, Backup.Error, Bytes.Bytes>, 82 cursor: Maybe<&1, VisibleCursor.Cursor>, 83 seen: Nat, 84 expected: Nat, 85 entry_count: U32, 86 block_count: U32, 87 next_index: U32, 88 next_phase: Phase 89 ) -> Step: 90 match result: 91 case Fail{error}: Failure{cursor, error} 92 case Done{bytes}: 93 Chunk{Stream{next_phase, cursor, Nil{}, 0n, seen, 94 expected, entry_count, block_count, next_index}, bytes} 95 96 def block.index.pair(index: U32) -> U32 & U32: 97 +current = U32.to_nat(index) 98 (U32.from_nat(current), U32.from_nat(Policy.block.index.next(current))) 99 100 def block.encoded( 101 cursor: Maybe<&1, VisibleCursor.Cursor>, 102 pending: List<&1, ByteMemTable.Entry>, 103 seen: Nat, 104 expected: Nat, 105 entry_count: U32, 106 block_count: U32, 107 indexes: U32 & U32, 108 count: Nat, 109 next_phase: Phase 110 ) -> Step: 111 match indexes: 112 case (encode_index, next_index): 113 block.written(Backup.block.encoded( 114 List.reverse(&1, ByteMemTable.Entry, pending), encode_index, 115 U32.from_nat(count)), cursor, seen, expected, 116 entry_count, block_count, next_index, next_phase) 117 118 def item.full( 119 cursor: Maybe<&1, VisibleCursor.Cursor>, 120 pending: List<&1, ByteMemTable.Entry>, 121 seen: Nat, 122 expected: Nat, 123 entry_count: U32, 124 block_count: U32, 125 block_index: U32, 126 count: Nat, 127 next_phase: Phase 128 ) -> Step: 129 block.encoded(cursor, pending, seen, expected, entry_count, block_count, 130 block.index.pair(block_index), count, next_phase) 131 132 def item.decided( 133 decision: Policy.AppendDecision, 134 cursor: Maybe<&1, VisibleCursor.Cursor>, 135 pending: List<&1, ByteMemTable.Entry>, 136 expected: Nat, 137 entry_count: U32, 138 block_count: U32, 139 block_index: U32 140 ) -> Step: 141 match decision: 142 case Policy.CursorRejected{}: 143 Failure{cursor, Backup.ResourceLimit{}} 144 case Policy.CursorBuffer{seen, pending_count}: 145 Continue{Stream{CursorBlocks{}, cursor, pending, pending_count, 146 seen, expected, entry_count, block_count, block_index}} 147 case Policy.CursorBlockReady{seen, pending_count}: 148 item.full(cursor, pending, seen, expected, entry_count, block_count, 149 block_index, pending_count, CursorBlocks{}) 150 151 def item.counted( 152 entry: ByteMemTable.Entry, 153 cursor: Maybe<&1, VisibleCursor.Cursor>, 154 pending: List<&1, ByteMemTable.Entry>, 155 seen: Nat, 156 pending_count: Nat, 157 entry_count: U32, 158 block_count: U32, 159 block_index: U32, 160 expected_copies: Nat & Nat 161 ) -> Step: 162 match expected_copies: 163 case (for_policy, retained): 164 item.decided(Policy.append(seen, pending_count, for_policy), 165 cursor, Con{entry, pending}, retained, entry_count, 166 block_count, block_index) 167 168 # Adds one cursor item without exceeding the declared total or block capacity. 169 def item( 170 entry: ByteMemTable.Entry, 171 cursor: Maybe<&1, VisibleCursor.Cursor>, 172 pending: List<&1, ByteMemTable.Entry>, 173 seen: Nat, 174 pending_count: Nat, 175 expected: Nat, 176 entry_count: U32, 177 block_count: U32, 178 block_index: U32 179 ) -> Step: 180 item.counted(entry, cursor, pending, seen, pending_count, 181 entry_count, block_count, block_index, 182 Backup.nat.duplicate.pair(expected)) 183 184 # Retains the pending block when the cursor advances over a tombstone. 185 def skip( 186 cursor: Maybe<&1, VisibleCursor.Cursor>, 187 pending: List<&1, ByteMemTable.Entry>, 188 pending_count: Nat, 189 seen: Nat, 190 expected: Nat, 191 entry_count: U32, 192 block_count: U32, 193 block_index: U32 194 ) -> Step: 195 Continue{Stream{CursorBlocks{}, cursor, pending, pending_count, seen, 196 expected, entry_count, block_count, block_index}} 197 198 def footer.ready( 199 cursor: Maybe<&1, VisibleCursor.Cursor>, 200 seen: Nat, 201 expected: Nat, 202 entry_count: U32, 203 block_count: U32, 204 block_index: U32, 205 encoded: Backup.Stream & Bytes.Bytes 206 ) -> Step: 207 match encoded: 208 case (_, bytes): 209 Chunk{Stream{CursorFinished{}, cursor, Nil{}, 0n, seen, 210 expected, entry_count, block_count, block_index}, bytes} 211 212 def footer.counts( 213 cursor: Maybe<&1, VisibleCursor.Cursor>, 214 seen: Nat, 215 expected: Nat, 216 entries: U32 & U32, 217 blocks: U32 & U32, 218 indexes: U32 & U32 219 ) -> Step: 220 match entries: 221 case (encoded_entries, retained_entries): 222 match blocks: 223 case (encoded_blocks, retained_blocks): 224 match indexes: 225 case (encoded_index, retained_index): 226 footer.ready(cursor, seen, expected, 227 retained_entries, retained_blocks, retained_index, 228 Backup.footer.bytes(Backup.Stream{Backup.Footer{}, Nil{}, 229 encoded_entries, encoded_blocks, encoded_index})) 230 231 # Emits the checksum footer while retaining only terminal cursor state. 232 def footer( 233 cursor: Maybe<&1, VisibleCursor.Cursor>, 234 seen: Nat, 235 expected: Nat, 236 entry_count: U32, 237 block_count: U32, 238 block_index: U32 239 ) -> Step: 240 footer.counts(cursor, seen, expected, 241 Backup.u32.duplicate.pair(entry_count), 242 Backup.u32.duplicate.pair(block_count), 243 Backup.u32.duplicate.pair(block_index)) 244 245 def finish.decided( 246 decision: Policy.FinishDecision, 247 cursor: Maybe<&1, VisibleCursor.Cursor>, 248 pending: List<&1, ByteMemTable.Entry>, 249 seen: Nat, 250 expected: Nat, 251 entry_count: U32, 252 block_count: U32, 253 block_index: U32 254 ) -> Step: 255 match decision: 256 case Policy.CursorFinishRejected{}: Failure{cursor, Backup.ResourceLimit{}} 257 case Policy.CursorFooter{}: 258 footer(cursor, seen, expected, entry_count, block_count, block_index) 259 case Policy.CursorFinalBlock{pending_count}: 260 block.encoded(cursor, pending, seen, expected, entry_count, 261 block_count, block.index.pair(block_index), pending_count, 262 CursorFooter{}) 263 264 def finish.copied( 265 cursor: Maybe<&1, VisibleCursor.Cursor>, 266 pending: List<&1, ByteMemTable.Entry>, 267 pending_count: Nat, 268 seen_copies: Nat & Nat, 269 expected_copies: Nat & Nat, 270 entry_count: U32, 271 block_count: U32, 272 block_index: U32 273 ) -> Step: 274 match seen_copies: 275 case (seen_check, seen_retained): 276 match expected_copies: 277 case (expected_check, expected_retained): 278 finish.decided(Policy.finish(seen_check, expected_check, pending_count), 279 cursor, pending, seen_retained, expected_retained, entry_count, 280 block_count, block_index) 281 282 # Rejects early cursor end; emits a final partial block before the footer. 283 def finish(stream: Stream) -> Step: 284 match stream: 285 case Stream{CursorBlocks{}, cursor, pending, pending_count, seen, expected, 286 entry_count, block_count, block_index}: 287 finish.copied(cursor, pending, pending_count, 288 Backup.nat.duplicate.pair(seen), 289 Backup.nat.duplicate.pair(expected), 290 entry_count, block_count, block_index) 291 case Stream{_, cursor, _, _, _, _, _, _, _}: Failure{cursor, Backup.CodecFailure{}} 292 293 def advance.active( 294 stream: Stream, 295 source_step: VisibleCursor.Step 296 ) -> Step: 297 match stream: 298 case Stream{CursorBlocks{}, cursor, pending, pending_count, seen, expected, 299 entry_count, block_count, block_index}: 300 match source_step: 301 case VisibleCursor.Item{entry, cursor}: 302 item(entry, Some{cursor}, pending, seen, pending_count, expected, 303 entry_count, block_count, block_index) 304 case VisibleCursor.Skip{cursor}: 305 skip(Some{cursor}, pending, pending_count, seen, expected, 306 entry_count, block_count, block_index) 307 case VisibleCursor.End{}: 308 finish(Stream{CursorBlocks{}, None{}, pending, pending_count, seen, 309 expected, entry_count, block_count, block_index}) 310 case VisibleCursor.Error{error}: 311 match error: 312 case (code, message): 313 Failure{None{}, Backup.SourceFailure{code, message}} 314 case Stream{CursorFooter{}, cursor, pending, pending_count, seen, 315 expected, entry_count, block_count, block_index}: 316 footer(cursor, seen, expected, entry_count, block_count, block_index) 317 case Stream{CursorFinished{}, _, _, _, _, _, _, _, _}: End{} 318 319 # Applies one result returned by the bounded visible cursor. 320 def advance(stream: Stream, source_step: VisibleCursor.Step) -> Step: 321 advance.active(stream, source_step) 322 323 # Applies one frame returned by a retained owner cursor. 324 def advance.owner.decision( 325 stream: Stream, 326 decision: Policy.OwnerFrameDecision 327 ) -> Step: 328 match stream: 329 case Stream{CursorBlocks{}, _, pending, pending_count, seen, expected, 330 entry_count, block_count, block_index}: 331 match decision: 332 case Policy.OwnerCursorItem{key, value}: 333 item(ByteMemTable.Entry{key, Some{value}}, None{}, pending, 334 seen, pending_count, expected, entry_count, block_count, 335 block_index) 336 case Policy.OwnerCursorEnd{}: 337 finish(Stream{CursorBlocks{}, None{}, pending, pending_count, 338 seen, expected, entry_count, block_count, block_index}) 339 case Stream{CursorFooter{}, cursor, pending, pending_count, seen, 340 expected, entry_count, block_count, block_index}: 341 footer(cursor, seen, expected, entry_count, block_count, block_index) 342 case Stream{CursorFinished{}, _, _, _, _, _, _, _, _}: End{} 343 344 def advance.owner(stream: Stream, frame: CursorCodec.Frame) -> Step: 345 advance.owner.decision(stream, Policy.owner.frame(frame))