1 import Base 2 import ./Decimal.bend as Decimal 3 import ./Keys.bend as Keys 4 import ./MemTable.bend as MemTable 5 import ./Sstable.bend as Sstable 6 import ./SstChecksum.bend as SstChecksum 7 8 # Compact SSTable v2: 9 # S2;;;# 10 # P;;; 11 # D;; 12 # The checksum is SstChecksum.digest(body) (hub SHA-256, hex). 13 14 def max_file_chars() -> Nat: 15 Nat.add(4294967295n, 1n) 16 17 def max_entries() -> Nat: 18 16777216n 19 20 def max_key_chars() -> Nat: 21 16777216n 22 23 def max_value_chars() -> Nat: 24 1073741824n 25 26 def max_level() -> Nat: 27 255n 28 29 def read_chunk_chars() -> Nat: 30 1048576n 31 32 type Error is Data: 33 UnknownVersion{} 34 MalformedDecimal{} 35 DecimalOverflow{} 36 InvalidLevel{} 37 EntryCountExceeded{} 38 InvalidEntryTag{} 39 KeyLengthExceeded{} 40 ValueLengthExceeded{} 41 TruncatedHeader{} 42 TruncatedEntry{} 43 TruncatedChecksum{} 44 UnsortedOrDuplicate{} 45 ChecksumMismatch{} 46 TrailingData{} 47 FileTooLarge{} 48 49 type Tag is Data: 50 PutTag{} 51 DelTag{} 52 53 type Phase is Data: 54 NeedS{} 55 Need2{} 56 NeedHeaderSemi{} 57 ReadLevel{scan: Decimal.Scan} 58 ReadCount{scan: Decimal.Scan} 59 ReadTag{} 60 ReadTagSemi{tag: Tag} 61 ReadKeyLength{tag: Tag, scan: Decimal.Scan} 62 ReadValueLength{key_length: Nat, scan: Decimal.Scan} 63 ReadKey{tag: Tag, remaining: Nat, value_length: Nat, reversed: String} 64 ReadValue{key: String, remaining: Nat, reversed: String} 65 NeedHash{} 66 ReadChecksum{reversed: String} 67 68 type Transition is Data: 69 Next{phase: Phase} 70 LevelRead{level: Nat} 71 CountRead{count: Nat} 72 EntryRead{entry: MemTable.Entry} 73 TransitionRejected{error: Error} 74 75 type Decoder is Data: 76 Dec{ 77 phase: Phase, 78 remaining_entries: Nat, 79 previous_key: Maybe<&2, String>, 80 reversed_entries: List<&2, MemTable.Entry>, 81 level: Nat, 82 declared_count: Nat, 83 body_rev: String, 84 consumed: Nat 85 } 86 DecoderRejected{error: Error} 87 88 type ParseResult is Data: 89 ParseRejected{error: Error} 90 Parsed{table: Sstable.Table} 91 92 # --- Serialization: reversed fragments followed by one String.join --- 93 94 def entry_fragments(entries: List<&2, MemTable.Entry>, acc: List<&2, String>) -> List<&2, String>: 95 match entries: 96 case Nil{}: 97 acc 98 case Con{MemTable.Entry{+key, +val}, rest}: 99 match val: 100 case None{}: 101 entry_fragments(rest, Con{key, Con{";", Con{Decimal.render(String.length(key)), Con{"D;", acc}}}}) 102 case Some{+value}: 103 entry_fragments(rest, Con{value, Con{key, Con{";", Con{Decimal.render(String.length(value)), Con{";", Con{Decimal.render(String.length(key)), Con{"P;", acc}}}}}}}) 104 105 def header_fragments(count: Nat, level: Nat) -> List<&2, String>: 106 Con{";", Con{Decimal.render(count), Con{";", Con{Decimal.render(level), Con{"S2;", Nil{}}}}}} 107 108 def body(+entries: List<&2, MemTable.Entry>, level: Nat) -> String: 109 +header_reversed = header_fragments(List.length(&2, MemTable.Entry, entries), level) 110 String.join(List.reverse(&2, String, entry_fragments(entries, header_reversed)), "") 111 112 def serialize(+entries: List<&2, MemTable.Entry>, level: Nat) -> String: 113 +encoded_body = body(entries, level) 114 encoded_body ++ "#" ++ SstChecksum.digest(encoded_body) 115 116 # --- Pure one-character decoder transitions --- 117 118 def expected_char(ok: Bool, next: Phase, error: Error) -> Transition: 119 match ok: 120 case True{}: 121 Next{next} 122 case False{}: 123 TransitionRejected{error} 124 125 def decimal_error(error: Decimal.Error, overflow: Error) -> Transition: 126 match error: 127 case Decimal.Overflow{}: 128 TransitionRejected{overflow} 129 case _: 130 TransitionRejected{MalformedDecimal{}} 131 132 def level_scan(scan: Decimal.Scan) -> Transition: 133 match scan: 134 case Decimal.Reading{value, started, leading_zero}: 135 Next{ReadLevel{Decimal.Reading{value, started, leading_zero}}} 136 case Decimal.Finished{value}: 137 LevelRead{value} 138 case Decimal.Failed{error}: 139 decimal_error(error, InvalidLevel{}) 140 141 def count_scan(scan: Decimal.Scan) -> Transition: 142 match scan: 143 case Decimal.Reading{value, started, leading_zero}: 144 Next{ReadCount{Decimal.Reading{value, started, leading_zero}}} 145 case Decimal.Finished{value}: 146 CountRead{value} 147 case Decimal.Failed{error}: 148 decimal_error(error, EntryCountExceeded{}) 149 150 def delete_key_length(value: Nat) -> Transition: 151 match value: 152 case 0n: 153 EntryRead{MemTable.Entry{"", None{}}} 154 case 1n+rest: 155 Next{ReadKey{DelTag{}, 1n+rest, 0n, ""}} 156 157 def key_length_done(tag: Tag, value: Nat) -> Transition: 158 match tag: 159 case PutTag{}: 160 Next{ReadValueLength{value, Decimal.Reading{0n, False{}, False{}}}} 161 case DelTag{}: 162 delete_key_length(value) 163 164 def key_length_scan(tag: Tag, scan: Decimal.Scan) -> Transition: 165 match scan: 166 case Decimal.Reading{value, started, leading_zero}: 167 Next{ReadKeyLength{tag, Decimal.Reading{value, started, leading_zero}}} 168 case Decimal.Failed{error}: 169 decimal_error(error, KeyLengthExceeded{}) 170 case Decimal.Finished{value}: 171 key_length_done(tag, value) 172 173 def put_lengths(key_length: Nat, value_length: Nat) -> Transition: 174 match key_length: 175 case 0n: 176 match value_length: 177 case 0n: 178 EntryRead{MemTable.Entry{"", Some{""}}} 179 case 1n+v: 180 Next{ReadValue{"", 1n+v, ""}} 181 case 1n+k: 182 Next{ReadKey{PutTag{}, 1n+k, value_length, ""}} 183 184 def value_length_scan(key_length: Nat, scan: Decimal.Scan) -> Transition: 185 match scan: 186 case Decimal.Reading{value, started, leading_zero}: 187 Next{ReadValueLength{key_length, Decimal.Reading{value, started, leading_zero}}} 188 case Decimal.Failed{error}: 189 decimal_error(error, ValueLengthExceeded{}) 190 case Decimal.Finished{value}: 191 put_lengths(key_length, value) 192 193 def tag_del(is_del: Bool) -> Transition: 194 match is_del: 195 case True{}: 196 Next{ReadTagSemi{DelTag{}}} 197 case False{}: 198 TransitionRejected{InvalidEntryTag{}} 199 200 def tag_put(is_put: Bool, ch: Char) -> Transition: 201 match is_put: 202 case True{}: 203 Next{ReadTagSemi{PutTag{}}} 204 case False{}: 205 tag_del(Char.is_eq(ch, 'D')) 206 207 def tag_step(+ch: Char) -> Transition: 208 tag_put(Char.is_eq(ch, 'P'), ch) 209 210 def key_last(tag: Tag, value_length: Nat, reversed: String) -> Transition: 211 match tag: 212 case DelTag{}: 213 EntryRead{MemTable.Entry{String.reverse(reversed), None{}}} 214 case PutTag{}: 215 match value_length: 216 case 0n: 217 EntryRead{MemTable.Entry{String.reverse(reversed), Some{""}}} 218 case 1n+v: 219 Next{ReadValue{String.reverse(reversed), 1n+v, ""}} 220 221 def key_step(tag: Tag, remaining: Nat, value_length: Nat, reversed: String, ch: Char) -> Transition: 222 match remaining: 223 case 0n: 224 TransitionRejected{TruncatedEntry{}} 225 case 1n+rest: 226 match rest: 227 case 0n: 228 key_last(tag, value_length, SCon{ch, reversed}) 229 case 1n+more: 230 Next{ReadKey{tag, 1n+more, value_length, SCon{ch, reversed}}} 231 232 def value_step(key: String, remaining: Nat, reversed: String, ch: Char) -> Transition: 233 match remaining: 234 case 0n: 235 TransitionRejected{TruncatedEntry{}} 236 case 1n+rest: 237 match rest: 238 case 0n: 239 EntryRead{MemTable.Entry{key, Some{String.reverse(SCon{ch, reversed})}}} 240 case 1n+more: 241 Next{ReadValue{key, 1n+more, SCon{ch, reversed}}} 242 243 def checksum_step(+reversed: String, +ch: Char) -> Transition: 244 Next{ReadChecksum{SCon{ch, reversed}}} 245 246 def phase_step(phase: Phase, +ch: Char) -> Transition: 247 match phase: 248 case NeedS{}: 249 expected_char(Char.is_eq(ch, 'S'), Need2{}, UnknownVersion{}) 250 case Need2{}: 251 expected_char(Char.is_eq(ch, '2'), NeedHeaderSemi{}, UnknownVersion{}) 252 case NeedHeaderSemi{}: 253 expected_char(Char.is_eq(ch, ';'), ReadLevel{Decimal.Reading{0n, False{}, False{}}}, UnknownVersion{}) 254 case ReadLevel{scan}: 255 level_scan(Decimal.scan_step(ch, scan, max_level(), Decimal.Delimited{';'})) 256 case ReadCount{scan}: 257 count_scan(Decimal.scan_step(ch, scan, max_entries(), Decimal.Delimited{';'})) 258 case ReadTag{}: 259 tag_step(ch) 260 case ReadTagSemi{tag}: 261 expected_char(Char.is_eq(ch, ';'), ReadKeyLength{tag, Decimal.Reading{0n, False{}, False{}}}, InvalidEntryTag{}) 262 case ReadKeyLength{tag, scan}: 263 key_length_scan(tag, Decimal.scan_step(ch, scan, max_key_chars(), Decimal.Delimited{';'})) 264 case ReadValueLength{key_length, scan}: 265 value_length_scan(key_length, Decimal.scan_step(ch, scan, max_value_chars(), Decimal.Delimited{';'})) 266 case ReadKey{tag, remaining, value_length, reversed}: 267 key_step(tag, remaining, value_length, reversed, ch) 268 case ReadValue{key, remaining, reversed}: 269 value_step(key, remaining, reversed, ch) 270 case NeedHash{}: 271 expected_char(Char.is_eq(ch, '#'), ReadChecksum{SNil{}}, TruncatedChecksum{}) 272 case ReadChecksum{reversed}: 273 checksum_step(reversed, ch) 274 275 def phase_bodies(phase: Phase) -> Bool: 276 match phase: 277 case NeedHash{}: 278 False{} 279 case ReadChecksum{reversed}: 280 False{} 281 case _: 282 True{} 283 284 def select_body(keeps: Bool, body_rev: String, ch: Char) -> String: 285 match keeps: 286 case True{}: 287 SCon{ch, body_rev} 288 case False{}: 289 body_rev 290 291 # --- Applying transitions enforces exact count and strict key order --- 292 293 def decoder_next(phase: Phase, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder: 294 Dec{phase, remaining, previous, entries, level, declared, body_rev, consumed} 295 296 def count_transition(count: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, body_rev: String, consumed: Nat) -> Decoder: 297 match count: 298 case 0n: 299 decoder_next(NeedHash{}, 0n, previous, entries, level, 0n, body_rev, consumed) 300 case 1n++rest: 301 decoder_next(ReadTag{}, 1n+rest, previous, entries, level, 1n+rest, body_rev, consumed) 302 303 def ordered_entry(entry: MemTable.Entry, remaining: Nat, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder: 304 match entry: 305 case MemTable.Entry{+key, +val}: 306 match remaining: 307 case 0n: 308 DecoderRejected{TrailingData{}} 309 case 1n+rest: 310 match rest: 311 case 0n: 312 decoder_next(NeedHash{}, 0n, Some{key}, Con{MemTable.Entry{key, val}, entries}, level, declared, body_rev, consumed) 313 case 1n+more: 314 decoder_next(ReadTag{}, 1n+more, Some{key}, Con{MemTable.Entry{key, val}, entries}, level, declared, body_rev, consumed) 315 316 def order_decision(cmp: Cmp, entry: MemTable.Entry, remaining: Nat, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder: 317 match cmp: 318 case LT{}: 319 ordered_entry(entry, remaining, entries, level, declared, body_rev, consumed) 320 case _: 321 DecoderRejected{UnsortedOrDuplicate{}} 322 323 def entry_transition(+entry: MemTable.Entry, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder: 324 match entry: 325 case MemTable.Entry{+key, +val}: 326 match previous: 327 case None{}: 328 ordered_entry(MemTable.Entry{key, val}, remaining, entries, level, declared, body_rev, consumed) 329 case Some{+old}: 330 order_decision(Keys.cmp(old, key), MemTable.Entry{key, val}, remaining, entries, level, declared, body_rev, consumed) 331 332 def apply_transition(transition: Transition, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat) -> Decoder: 333 match transition: 334 case TransitionRejected{error}: 335 DecoderRejected{error} 336 case Next{phase}: 337 decoder_next(phase, remaining, previous, entries, level, declared, body_rev, consumed) 338 case LevelRead{new_level}: 339 decoder_next(ReadCount{Decimal.Reading{0n, False{}, False{}}}, remaining, previous, entries, new_level, declared, body_rev, consumed) 340 case CountRead{count}: 341 count_transition(count, previous, entries, level, body_rev, consumed) 342 case EntryRead{entry}: 343 entry_transition(entry, remaining, previous, entries, level, declared, body_rev, consumed) 344 345 def decoder_step_live(+phase: Phase, remaining: Nat, previous: Maybe<&2, String>, entries: List<&2, MemTable.Entry>, level: Nat, declared: Nat, body_rev: String, consumed: Nat, +ch: Char) -> Decoder: 346 +next_body = select_body(phase_bodies(phase), body_rev, ch) 347 apply_transition(phase_step(phase, ch), remaining, previous, entries, level, declared, next_body, Nat.add(consumed, 1n)) 348 349 def decoder_step_bounded( 350 within: Bool, 351 phase: Phase, 352 remaining: Nat, 353 previous: Maybe<&2, String>, 354 entries: List<&2, MemTable.Entry>, 355 level: Nat, 356 declared: Nat, 357 body_rev: String, 358 consumed: Nat, 359 ch: Char, 360 ) -> Decoder: 361 match within: 362 case False{}: 363 DecoderRejected{FileTooLarge{}} 364 case True{}: 365 decoder_step_live(phase, remaining, previous, entries, level, declared, body_rev, consumed, ch) 366 367 def decoder_step(decoder: Decoder, ch: Char) -> Decoder: 368 match decoder: 369 case DecoderRejected{error}: 370 DecoderRejected{error} 371 case Dec{phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, +consumed}: 372 decoder_step_bounded(Nat.is_lt(consumed, max_file_chars()), phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, consumed, ch) 373 374 # --- EOF validation and table construction --- 375 376 def checksum_decision(equal: Bool, entries: List<&2, MemTable.Entry>, level: Nat) -> ParseResult: 377 match equal: 378 case False{}: 379 ParseRejected{ChecksumMismatch{}} 380 case True{}: 381 Parsed{Sstable.from_sorted_unique(List.reverse(&2, MemTable.Entry, entries), level)} 382 383 def checksum_value(+claimed_rev: String, entries: List<&2, MemTable.Entry>, level: Nat, +body_rev: String) -> ParseResult: 384 checksum_decision(SstChecksum.verify(String.reverse(body_rev), String.reverse(claimed_rev)), entries, level) 385 386 def finish_decoder(decoder: Decoder) -> ParseResult: 387 match decoder: 388 case DecoderRejected{error}: 389 ParseRejected{error} 390 case Dec{phase, remaining_entries, previous_key, reversed_entries, level, declared_count, body_rev, consumed}: 391 match phase: 392 case ReadChecksum{reversed}: 393 checksum_value(reversed, reversed_entries, level, body_rev) 394 case NeedS{}: 395 ParseRejected{TruncatedHeader{}} 396 case Need2{}: 397 ParseRejected{TruncatedHeader{}} 398 case NeedHeaderSemi{}: 399 ParseRejected{TruncatedHeader{}} 400 case ReadLevel{scan}: 401 ParseRejected{TruncatedHeader{}} 402 case ReadCount{scan}: 403 ParseRejected{TruncatedHeader{}} 404 case NeedHash{}: 405 ParseRejected{TruncatedChecksum{}} 406 case _: 407 ParseRejected{TruncatedEntry{}} 408 409 def decode_go(rest: String, decoder: Decoder) -> ParseResult: 410 match rest: 411 case SNil{}: 412 finish_decoder(decoder) 413 case SCon{c, t}: 414 decode_go(t, decoder_step(decoder, c)) 415 416 def parse_result(str: String) -> ParseResult: 417 decode_go(str, Dec{NeedS{}, 0n, None{}, Nil{}, 0n, 0n, "", 0n}) 418 419 def parse_wrap(result: ParseResult) -> Maybe<&2, Sstable.Table>: 420 match result: 421 case ParseRejected{error}: 422 None{} 423 case Parsed{table}: 424 Some{table} 425 426 def parse(str: String) -> Maybe<&2, Sstable.Table>: 427 parse_wrap(parse_result(str)) 428 429 # Useful pure projections for laws and streaming-state agreement. 430 def parser_decides(_result: ParseResult) -> Bool: 431 True{} 432 433 def error_of(result: ParseResult) -> Maybe<&2, Error>: 434 match result: 435 case ParseRejected{error}: 436 Some{error} 437 case Parsed{table}: 438 None{} 439 440 # --- Fail-closed batch checksum verdict (Task 3) --- 441 # Any parse failure (checksum or otherwise) is a bad body: recovery treats 442 # every reject as fatal, so the batch verdict stays consistent with parse. 443 def checksum_bad(err: Maybe<&2, Error>) -> Bool: 444 match err: 445 case None{}: 446 False{} 447 case Some{_}: 448 True{} 449 450 # Invert a bad-body flag into an ok verdict. 451 def checksum_ok_of(bad: Bool) -> Bool: 452 match bad: 453 case True{}: 454 False{} 455 case False{}: 456 True{} 457 458 # Checksum verdict over a full encoded body, reusing the production decoder. 459 def body_checksum_ok(+enc: String) -> Bool: 460 checksum_ok_of(checksum_bad(error_of(parse_result(enc))))