Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 | 25x 25x 65990x 25x 11615x 25x 3481x 1818x 1663x 9x 1654x 25x 25x 25673x 25x 2463x 25x 616x 25x 18337x 49869x 37085x 18337x 18337x 49869x 49869x 35833x 35833x 14036x 18337x 25x 3360x 3360x 3477x 3477x 3477x 1156x 590x 566x 320x 246x 53x 246x 246x 98x 3477x 3376x 3360x 25x 4286x 4286x 4415x 1220x 3195x 4286x 25x 8176x 8176x 8176x 2895x 2111x 784x 8176x 8176x 25x 5203x 528x 5203x 576x 5203x 25x 2973x 1741x 1232x 1232x 1232x 1232x 140x 1232x 1232x 25x 12342x 12342x 12417x 189557x 189557x 189504x 189557x 12342x 25x 12343x 12343x 12344x 12344x 12344x 2x 12342x 12342x 12342x 12343x 12343x 12343x 25x 5203x 5203x 5203x 5203x 5203x 1957x 1957x 3246x 3246x 3246x 3246x 3246x 3246x 393x 3246x 3246x 3246x 3246x 3246x 5203x 5203x 25x 5203x 5203x 5203x 5203x 5203x 5203x 5203x 5203x 5203x 5203x 5203x 25x 982x 982x 982x 982x 982x 25x 1832x 1832x 1832x 1832x 1832x 1832x 1x 1832x 264x 1832x 633x 633x 633x 25x 633x 1199x 1199x 1199x 1199x 1x 1199x 25x 5156x 5156x 2x 5154x 1x 5153x 5153x 11x 5153x 5156x 576x 576x 562x 5153x 11x 5153x 25x 710x 710x 331x 379x 379x 48x 331x 331x 331x 331x 331x 319x 331x 25x 589x 589x 4x 585x 585x 585x 585x 585x 585x 279x 279x 585x 103x 585x 25x 1143x 1143x 1143x 1143x 1143x 2x 1141x 1141x 6x 6x 1141x 1319x 1141x 18x 18x 1143x 25x 8x 8x 8x 8x 8x 8x 25x 20529x 12343x 5203x 1832x 1133x 10x 10x 10x 10x 3x 10x 8x 8x 25x 3421x 3421x 3421x 3421x 3421x 61x 61x 3360x 3360x 378x 378x 2982x 2982x 2982x 3421x 3421x 1902x 1902x 1080x 25x 10920x 10920x 838x 10920x 25x 11240x 4x 4x 11240x 763x 161x 763x 11240x 11240x 11240x 25x 11240x 11240x 11240x 25x 14753x 14753x 14753x 14753x 14753x 14753x 14753x 14753x 4507x 1178x 14753x 25x 4529x 4529x 1658x 4529x 4529x 4529x 6172x 4529x 1x 4529x 25x 340x 271x 69x 25x 452x 502x 166x 286x 25x 406x 406x 406x 406x 680x 406x 406x 406x 405x 25x 1554x 25x 89853x 89853x 10897x 78956x 4x 4x 3x 1x 78952x 11615x 11615x 22575x 14753x 1127x 1061x 1061x 1x 1060x 3517x 3517x 3517x 267x 267x 212x 212x 174x 1464x 44493x 20437x 20437x 2294x 5141x 5141x 704x 704x 10920x 10920x 856x 188x 188x 314x 314x 3472x 3x 3469x 166x 1x 267x 92x 92x 7x 85x 15x 15x 6x 6x 2x 4x 8x 8x 2x 6x 43x 10x 6x 6x 4x 2x 87x 2x | import type { Timestamp } from '@bufbuild/protobuf/wkt'
import { StatusIds_StatusCode } from '@ydbjs/api/operation'
import { loggers } from '@ydbjs/debug'
import { YDBError } from '@ydbjs/error'
import type { TransitionResult, TransitionRuntime } from '@ydbjs/fsm'
import { isRetryableError, isRetryableStreamError } from '@ydbjs/retry'
import { TopicPartitionSession } from '../partition-session.js'
// The pure half of the reader: states, context, and a synchronous transition
// with no I/O. Mirrors writer-state.ts. The transport FSM owns one streamRead
// stream and forwards server frames; the runtime's mapTransportOutput CLASSIFIES
// them into the typed `reader.stream.*` events below, so this transition works on
// clean domain events — never raw protobuf.
//
// The full transition map (table + diagram) lives in packages/topic/ARCHITECTURE.md —
// update it in the same commit when you change this dispatch.
//
// KEY DESIGN (see project decisions): all reader state is keyed by the STABLE
// (topicPath, partitionId) pair — partitionKey() — never the ephemeral
// partitionSessionId (proto: "unique inside one RPC call"). partition_id alone is
// per-topic, so a multi-topic reader routinely sees the same id from several topics.
// Pending commits are held across reconnect and reconciled per-partition on each
// start_partition, so commit() is never rejected by a transparent reconnect.
// Commit promises live in the facade keyed by waiterId — the transition holds no
// callbacks, which keeps it model-testable.
let dbg = loggers.topic.extend('reader')
// ── State / context ─────────────────────────────────────────────────────────────
export type ReaderState =
| 'idle'
| 'connecting'
| 'ready'
| 'reconnecting'
| 'closing'
| 'closed'
| 'errored'
// Half-open [start, end) offset interval — the only range shape on the wire.
export type OffsetRange = { start: bigint; end: bigint }
// Stable identity of a partition across reconnects. '|' cannot appear in a YDB
// scheme path, so the key is collision-free.
export let partitionKey = function partitionKey(path: string, partitionId: bigint): string {
return `${path}|${partitionId}`
}
// A commit awaiting the server's high-water-mark ack. `targetOffset` controls promise
// completion; `wireRanges` is only the unshared coverage this commit put (or will put)
// on the wire. A fully claimed duplicate has no wire ranges but still waits for the
// server watermark instead of resolving optimistically.
export type PendingCommit = {
targetOffset: bigint
wireRanges: OffsetRange[]
waiterId: number
}
// Everything the reader tracks about one partition, keyed by the stable partitionKey.
export type PartitionEntry = {
// identity — stable across reconnects
partitionId: bigint
path: string
// current ephemeral grant (replaced on each reconnect/reassign)
partitionSessionId: bigint
session: TopicPartitionSession
// Identity of the latest grant (start_partition), stamped through the async
// onPartitionSessionStart handshake: partition session ids restart at 1 per
// stream, so without it a hook completing after a reconnect could answer a NEW
// grant that reused the id — the server kills the session for a double response.
grantId: number
// True from the grant until its start_ready is honored. While set, commits are
// buffered (the ack performs the single send — a commit must never hit the wire
// twice, nor before the start response), and the reassign gc treats the entry as
// not-yet-live so a hung hook cannot strand commit() waiters forever.
ackPending: boolean
// offset tracking
partitionOffsets: OffsetRange
partitionCommittedOffset: bigint
// A start response requests this offset but does not acknowledge its durability.
requestedCommitOffset: bigint | undefined
// Delivery watermark: end of the last delivered message's stitched commit range.
// Reset to committed_offset on every grant (redelivery restarts there). Each
// delivered message's commit range starts here, so server-side offset holes
// (retention, readFrom skips) are attributed to the NEXT delivered message —
// and a delivered-but-unacked message is never covered by another message's range.
deliveredWatermark: bigint
// Offsets already claimed by a sent/buffered commit. Re-sending an overlapping
// or rewound range is session-fatal (BAD_REQUEST "double committing is
// forbidden"), so every new commit subtracts this coverage first. Rebuilt from
// the narrowed pending commits on each grant; compacted as the server watermark
// advances.
claimedRanges: OffsetRange[]
// lifecycle
state: 'active' | 'stopping-graceful' | 'stopped' | 'ended'
// Graceful-stop handshake: set once the async onPartitionSessionStop hook
// completed; the stop response goes out only when stopReady AND the pending
// commits drained (either order), or the per-partition timeout escalates.
stopReady: boolean
pendingCommits: PendingCommit[]
}
// One partition's slice of a ReadResponse — the structural shape the transition
// consumes (a subset of the protobuf PartitionData).
export type PartitionReadData = {
partitionSessionId: bigint
batches: {
producerId: string
codec: number
// Optional protobuf message fields are `T | undefined` (protoc-gen-es honours
// exactOptionalPropertyTypes): present-but-unset must be assignable here.
writtenAt?: Timestamp | undefined
messageData: {
offset: bigint
seqNo: bigint
data: Uint8Array
uncompressedSize: bigint
createdAt?: Timestamp | undefined
metadataItems: { key: string; value: Uint8Array }[]
}[]
}[]
}
// A message parsed out of a ReadResponse. The transition emits these; the runtime
// decompresses `data` via the codec map and builds the user-facing TopicMessage.
export type ReaderMessage = {
offset: bigint
seqNo: bigint
data: Uint8Array
uncompressedSize: bigint
producer: string
codec: number
// Start of this message's commit range (stitched at delivery time — covers the
// server-side offset hole immediately preceding the message, if any).
commitRangeStart: bigint
createdAt?: Timestamp
writtenAt?: Timestamp
metadataItems: { key: string; value: Uint8Array }[]
}
export type ReaderLimits = {
maxBufferBytes: bigint
}
// Pure logical context — mutated synchronously inside the transition only.
export type ReaderCtx = {
// connection identity
sessionId: string
hasEverConnected: boolean
// reconnect bookkeeping (mirrors the writer)
attempts: number
lastError: unknown
// When set, a SCHEME_ERROR (e.g. the topic does not exist yet) is retried instead
// of being fatal — the reader waits until the topic is created.
retryOnSchemeError: boolean
// Terminal reconnect deadline (ms). Infinity = unbounded (reconnect forever); the
// transition owns whether to arm the `recovery_window` timer based on this.
recoveryWindowMs: number
// partition registry — keyed by the STABLE partitionKey (topicPath + partitionId)
partitions: Map<string, PartitionEntry>
// ephemeral partitionSessionId -> partitionKey, rebuilt each stream
sessionIndex: Map<bigint, string>
// monotonic grant counter — source of PartitionEntry.grantId
grantSeq: number
// byte flow-control
inFlightBytes: bigint
limits: ReaderLimits
}
export type GlobalTimerName =
| 'start_timeout'
| 'retry_backoff'
| 'recovery_window'
| 'update_token'
// The close() drain deadline armed by toClosing.
| 'graceful_timeout'
// Per-partition timers — the `partition_` prefix IS the scoping marker: their
// effects/events carry a required partitionKey and key as `<name>:<partitionKey>`.
export type PartitionTimerName =
| 'partition_graceful_timeout'
| 'partition_reassign_gc'
| 'partition_commit_status'
export type TimerName = GlobalTimerName | PartitionTimerName
// Timer control payload — partitionKey is required exactly when the name is partition_*.
export type TimerRef =
| { which: GlobalTimerName }
| { which: PartitionTimerName; partitionKey: string }
// Typed server-stream events, classified from the protobuf in mapTransportOutput.
export type ReaderStreamEvent =
| { type: 'reader.stream.read_response'; partitionData: PartitionReadData[]; bytesSize: bigint }
| {
type: 'reader.stream.partition_status'
partitionSessionId: bigint
committedOffset: bigint
}
| {
type: 'reader.stream.start_partition'
partitionSessionId: bigint
partitionId: bigint
path: string
committedOffset: bigint
partitionOffsets: OffsetRange
}
| {
type: 'reader.stream.stop_partition'
partitionSessionId: bigint
graceful: boolean
committedOffset: bigint
}
| {
type: 'reader.stream.commit_response'
committed: { partitionSessionId: bigint; committedOffset: bigint }[]
}
| {
type: 'reader.stream.end_partition'
partitionSessionId: bigint
childPartitionIds: bigint[]
adjacentPartitionIds: bigint[]
}
export type ReaderEvent =
// user (dispatched by the facade)
| { type: 'reader.start' }
// The facade builds the merged half-open ranges from the messages' stitched
// commit-range starts; the transition clamps them against server truth.
| { type: 'reader.commit'; partitionKey: string; ranges: OffsetRange[]; waiterId: number }
| { type: 'reader.read_release'; bytes: bigint }
| { type: 'reader.close' }
| { type: 'reader.destroy'; reason?: unknown }
// transport -> reader
| { type: 'reader.stream.init_response'; sessionId: string }
| ReaderStreamEvent
| { type: 'reader.stream.disconnected'; error?: unknown }
// The async onPartitionSessionStart hook finished (runs detached in the runtime so
// user code never blocks the drain loop); carries the hook's offset overrides and
// the grantId of the grant it answers (guards against stale completions).
| {
type: 'reader.partition.start_ready'
partitionSessionId: bigint
partitionKey: string
grantId: number
readOffset?: bigint
commitOffset?: bigint
}
// The async onPartitionSessionStop hook finished for a graceful stop — the stop
// response may go out once the pending commits drain too.
| {
type: 'reader.partition.stop_ready'
partitionKey: string
grantId: number
}
// timers
| { type: 'reader.timer.start_timeout' }
| { type: 'reader.timer.retry_backoff' }
| { type: 'reader.timer.recovery_window' }
| { type: 'reader.timer.update_token' }
// The close() drain deadline armed by toClosing.
| { type: 'reader.timer.graceful_timeout' }
// Per-partition graceful-stop fallback.
| { type: 'reader.timer.partition_graceful_timeout'; partitionKey: string }
| { type: 'reader.timer.partition_reassign_gc'; partitionKey: string }
| { type: 'reader.timer.partition_commit_status'; partitionKey: string }
export type ReaderEffect =
| { type: 'reader.effect.transport.connect' }
// Domain-level client messages; the runtime builds the protobuf wire frame (mirrors
// the writer, whose transition emits MessageData and the runtime frames the request).
| { type: 'reader.effect.send.read_request'; bytesSize: bigint }
| {
type: 'reader.effect.send.commit'
partitionSessionId: bigint
ranges: OffsetRange[]
}
| { type: 'reader.effect.send.stop_response'; partitionSessionId: bigint }
| { type: 'reader.effect.send.partition_status'; partitionSessionId: bigint }
| {
type: 'reader.effect.send.start_response'
partitionSessionId: bigint
readOffset?: bigint
commitOffset?: bigint
}
| { type: 'reader.effect.send.update_token' }
| { type: 'reader.effect.transport.close' }
// Runs the async onPartitionSessionStart callback then sends the response.
| {
type: 'reader.effect.partition.start_hook'
partitionSessionId: bigint
partitionKey: string
grantId: number
committedOffset: bigint
partitionOffsets: OffsetRange
}
// Runs the async onPartitionSessionStop callback for a graceful stop, then
// dispatches reader.partition.stop_ready. The session is still committable while
// the hook runs — this is the documented "last chance to commit".
| {
type: 'reader.effect.partition.stop_hook'
partitionSessionId: bigint
partitionKey: string
grantId: number
committedOffset: bigint
}
| ({ type: 'reader.effect.timer.schedule' } & TimerRef)
| ({ type: 'reader.effect.timer.clear' } & TimerRef)
| { type: 'reader.effect.finalize'; reason: unknown }
export type ReaderOutput =
| { type: 'reader.session'; sessionId: string }
// One output per ReadResponse. `releaseBytes` is the whole response size; the
// facade dispatches reader.read_release with it once — a response spanning several
// partitions releases its credit exactly once, not per partition.
| {
type: 'reader.messages'
releaseBytes: bigint
groups: { session: TopicPartitionSession; messages: ReaderMessage[] }[]
}
| {
type: 'reader.partition.started'
partitionId: bigint
partitionSessionId: bigint
committedOffset: bigint
session: TopicPartitionSession
}
| {
type: 'reader.partition.stopped'
partitionId: bigint
reason: 'graceful' | 'lost' | 'ended'
session: TopicPartitionSession
}
| {
type: 'reader.partition.committed'
partitionId: bigint
committedOffset: bigint
session: TopicPartitionSession
}
| { type: 'reader.commit.resolved'; waiterId: number }
| { type: 'reader.commit.rejected'; waiterId: number; reason: unknown }
| { type: 'reader.reconnecting'; attempt: number; error?: unknown }
| { type: 'reader.error'; error: unknown }
| { type: 'reader.closed'; reason?: unknown }
type ReaderRuntime = TransitionRuntime<ReaderState, ReaderEvent, ReaderOutput>
// ── Helpers ─────────────────────────────────────────────────────────────────────
export let createReaderCtx = function createReaderCtx(
limits: ReaderLimits,
options?: { retryOnSchemeError?: boolean; recoveryWindowMs?: number }
): ReaderCtx {
return {
sessionId: '',
hasEverConnected: false,
attempts: 0,
lastError: undefined,
retryOnSchemeError: options?.retryOnSchemeError ?? false,
recoveryWindowMs: options?.recoveryWindowMs ?? Infinity,
partitions: new Map(),
sessionIndex: new Map(),
grantSeq: 0,
inFlightBytes: 0n,
limits,
}
}
// Reconnecting is always safe for a reader (offsets are server-tracked), so we
// retry any retryable stream error; a clean end (undefined) is a server-side
// reconnect. Fatal statuses (SCHEME_ERROR, UNAUTHORIZED, …) stop the reader —
// except SCHEME_ERROR is retried when `retryOnSchemeError` is set (wait for the
// topic to be created).
export let isRetryableReaderError = function isRetryableReaderError(
error: unknown,
retryOnSchemeError = false
): boolean {
if (error === undefined || error === null) {
return true
}
if (
retryOnSchemeError &&
error instanceof YDBError &&
error.code === StatusIds_StatusCode.SCHEME_ERROR
) {
return true
}
return isRetryableStreamError(error) || isRetryableError(error, false)
}
let clearConnectTimersEffects: ReaderEffect[] = [
{ type: 'reader.effect.timer.clear', which: 'start_timeout' },
{ type: 'reader.effect.timer.clear', which: 'retry_backoff' },
{ type: 'reader.effect.timer.clear', which: 'recovery_window' },
]
let readRequestEffect = function readRequestEffect(bytesSize: bigint): ReaderEffect {
return { type: 'reader.effect.send.read_request', bytesSize }
}
let commitEffect = function commitEffect(
partitionSessionId: bigint,
ranges: OffsetRange[]
): ReaderEffect {
return { type: 'reader.effect.send.commit', partitionSessionId, ranges }
}
let stopResponseEffect = function stopResponseEffect(partitionSessionId: bigint): ReaderEffect {
return { type: 'reader.effect.send.stop_response', partitionSessionId }
}
// ── Range algebra ───────────────────────────────────────────────────────────────
// All lists are kept sorted by start with no overlapping or touching neighbours.
// Merge a batch of ranges into normalized form (sort, join overlapping/adjacent).
export let mergeRanges = function mergeRanges(ranges: OffsetRange[]): OffsetRange[] {
let sorted = ranges
.filter((r) => r.end > r.start)
.sort((a, b) => (a.start < b.start ? -1 : a.start > b.start ? 1 : 0))
let merged: OffsetRange[] = []
for (let range of sorted) {
let last = merged[merged.length - 1]
if (last !== undefined && range.start <= last.end) {
Eif (range.end > last.end) {
last.end = range.end
}
} else {
merged.push({ start: range.start, end: range.end })
}
}
return merged
}
// Subtract `coverage` from `ranges` and clamp everything below `floor` — the
// remainder is what may still go on the wire without double-committing.
let subtractCovered = function subtractCovered(
ranges: OffsetRange[],
coverage: OffsetRange[],
floor: bigint
): OffsetRange[] {
let remainder: OffsetRange[] = []
for (let range of ranges) {
let start = range.start < floor ? floor : range.start
let end = range.end
for (let covered of coverage) {
if (covered.end <= start) {
continue
}
if (covered.start >= end) {
break
}
if (covered.start > start) {
remainder.push({ start, end: covered.start })
}
start = covered.end
if (start >= end) {
break
}
}
if (start < end) {
remainder.push({ start, end })
}
}
return mergeRanges(remainder)
}
// Drop coverage the server already confirmed (everything below the watermark).
let compactCoverage = function compactCoverage(
coverage: OffsetRange[],
watermark: bigint
): OffsetRange[] {
let compacted: OffsetRange[] = []
for (let range of coverage) {
if (range.end <= watermark) {
continue
}
compacted.push({
start: range.start < watermark ? watermark : range.start,
end: range.end,
})
}
return compacted
}
// Resolve pending commits already covered by a committed high-water mark; return the
// waiterIds to resolve. Used by both the commit-ack path and the reconnect reconcile.
let drainCommits = function drainCommits(entry: PartitionEntry, committedOffset: bigint): number[] {
let resolved: number[] = []
let kept: PendingCommit[] = []
for (let pending of entry.pendingCommits) {
if (pending.targetOffset <= committedOffset) {
resolved.push(pending.waiterId)
} else {
kept.push(pending)
}
}
entry.pendingCommits = kept
return resolved
}
// Re-grant reconcile: narrow every pending's wire ranges to server truth, resolve by
// target watermark, and rebuild the claimed coverage from what is left. Empty wire
// ranges remain pending when another in-flight commit owns their coverage.
let narrowPendings = function narrowPendings(
entry: PartitionEntry,
committedOffset: bigint,
runtime: ReaderRuntime
): void {
for (let waiterId of drainCommits(entry, committedOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
for (let pending of entry.pendingCommits) {
pending.wireRanges = compactCoverage(pending.wireRanges, committedOffset)
}
entry.claimedRanges = mergeRanges(entry.pendingCommits.flatMap((pending) => pending.wireRanges))
}
// Advance the server-confirmed committed watermark; emits partition.committed (the
// onCommittedOffset observer must see every advance — commit acks, stop_partition
// watermarks, and partition status responses alike). Returns true when it advanced.
let advanceCommitted = function advanceCommitted(
entry: PartitionEntry,
committedOffset: bigint,
runtime: ReaderRuntime
): boolean {
if (committedOffset <= entry.partitionCommittedOffset) {
return false
}
entry.partitionCommittedOffset = committedOffset
entry.session.partitionCommittedOffset = committedOffset
entry.claimedRanges = compactCoverage(entry.claimedRanges, committedOffset)
if (entry.deliveredWatermark < committedOffset) {
entry.deliveredWatermark = committedOffset
}
runtime.emit({
type: 'reader.partition.committed',
partitionId: entry.partitionId,
committedOffset,
session: entry.session,
})
return true
}
// Decode one ReadResponse partition into messages, stitching each message's commit
// range start from the delivery watermark: server-side offset holes are attributed
// to the next delivered message, never to a later commit of an unrelated one. Pure
// w.r.t. flow-control (the caller charges bytes once per response). The FSM is
// tx-agnostic: tx read-offset tracking lives in the facade, which sees every
// delivered message.
let collectMessages = function collectMessages(
entry: PartitionEntry,
partitionData: PartitionReadData
): ReaderMessage[] {
let messages: ReaderMessage[] = []
for (let batch of partitionData.batches) {
for (let md of batch.messageData) {
let commitRangeStart =
entry.deliveredWatermark < md.offset ? entry.deliveredWatermark : md.offset
if (entry.deliveredWatermark < md.offset + 1n) {
entry.deliveredWatermark = md.offset + 1n
}
messages.push({
offset: md.offset,
seqNo: md.seqNo,
data: md.data,
uncompressedSize: md.uncompressedSize,
producer: batch.producerId,
codec: batch.codec,
commitRangeStart,
...(md.createdAt && { createdAt: md.createdAt }),
...(batch.writtenAt && { writtenAt: batch.writtenAt }),
metadataItems: md.metadataItems,
})
}
}
return messages
}
// ── Server-stream event handlers ─────────────────────────────────────────────────
let readResponse = function readResponse(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.read_response' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let groups: { session: TopicPartitionSession; messages: ReaderMessage[] }[] = []
for (let partitionData of event.partitionData) {
let key = ctx.sessionIndex.get(partitionData.partitionSessionId)
let entry = key !== undefined ? ctx.partitions.get(key) : undefined
// Drop data for an unknown, superseded (stale session id after a reassign), or
// no-longer-active partition. A gracefully stopping partition still delivers:
// the server sends no NEW data after the stop request, but in-flight responses
// belong to the app — the soft-stop contract is "finish processing and commit".
if (
!entry ||
entry.partitionSessionId !== partitionData.partitionSessionId ||
entry.state === 'stopped' ||
entry.state === 'ended'
) {
continue
}
let messages = collectMessages(entry, partitionData)
Eif (messages.length > 0) {
groups.push({ session: entry.session, messages })
}
}
// Charge the whole response once and emit a single batch so the consumer releases
// exactly bytesSize — several partitions in one response must not each claim the
// full size. Emit even when everything was dropped so credit is still released.
ctx.inFlightBytes += event.bytesSize
runtime.emit({ type: 'reader.messages', releaseBytes: event.bytesSize, groups })
return []
}
// Create or refresh the registry entry for a (re)started partition, installing the
// fresh ephemeral grant. Pending commits survive (narrowed to server truth); the
// delivery watermark and claimed coverage restart from the committed offset.
let upsertPartitionEntry = function upsertPartitionEntry(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.start_partition' }>,
session: TopicPartitionSession
): PartitionEntry {
let { partitionSessionId, partitionId, path, committedOffset, partitionOffsets } = event
let key = partitionKey(path, partitionId)
let entry = ctx.partitions.get(key)
ctx.grantSeq += 1
if (entry === undefined) {
entry = {
partitionId,
path,
partitionSessionId,
session,
grantId: ctx.grantSeq,
ackPending: true,
partitionOffsets,
partitionCommittedOffset: committedOffset,
requestedCommitOffset: undefined,
deliveredWatermark: committedOffset,
claimedRanges: [],
state: 'active',
stopReady: false,
pendingCommits: [],
}
ctx.partitions.set(key, entry)
} else {
// Reconnect/reassign: drop the superseded session id from the index (so a late
// message on the dead session can never misroute), install the fresh session
// object + id, keep pending commits for the reconcile in startPartitionSession.
ctx.sessionIndex.delete(entry.partitionSessionId)
entry.session = session
entry.partitionSessionId = partitionSessionId
entry.partitionOffsets = partitionOffsets
entry.requestedCommitOffset = undefined
// Server truth may only move the committed mark forward.
if (entry.partitionCommittedOffset < committedOffset) {
entry.partitionCommittedOffset = committedOffset
}
// Redelivery restarts at the committed offset — stitch from there again.
entry.deliveredWatermark = entry.partitionCommittedOffset
entry.state = 'active'
entry.stopReady = false
entry.grantId = ctx.grantSeq
entry.ackPending = true
}
ctx.sessionIndex.set(partitionSessionId, key)
return entry
}
let startPartitionSession = function startPartitionSession(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.start_partition' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let { partitionSessionId, partitionId, path, committedOffset, partitionOffsets } = event
let key = partitionKey(path, partitionId)
let session = new TopicPartitionSession(partitionSessionId, partitionId, path)
session.partitionOffsets = partitionOffsets
session.partitionCommittedOffset = committedOffset
let entry = upsertPartitionEntry(ctx, event, session)
let effects: ReaderEffect[] = [
{ type: 'reader.effect.timer.clear', which: 'partition_commit_status', partitionKey: key },
// A stale graceful-stop fallback from the previous stream must not fire against
// the freshly granted session.
{
type: 'reader.effect.timer.clear',
which: 'partition_graceful_timeout',
partitionKey: key,
},
]
// COMMIT RECONCILE, half one: resolve pendings covered by committed_offset and
// narrow the remainder ranges to [committed_offset, …). The re-send happens in
// ackPartitionStart — after the start response — so a commit can never reach the
// wire before the response that makes the session fully live (the reassign gc
// also stays armed until then, bounding a hung onPartitionSessionStart hook).
narrowPendings(entry, entry.partitionCommittedOffset, runtime)
runtime.emit({
type: 'reader.partition.started',
partitionId,
partitionSessionId,
committedOffset,
session,
})
// The response's read/commit offsets may be overridden by the async
// onPartitionSessionStart callback, so the runtime sends it from an effect.
effects.push({
type: 'reader.effect.partition.start_hook',
partitionSessionId,
partitionKey: key,
grantId: entry.grantId,
committedOffset,
partitionOffsets,
})
return effects
}
// Retire a partition session: mark it stopped, drop it from the ephemeral index,
// and notify the facade. Pending commits stay on the entry for reconcile/gc.
let markStopped = function markStopped(
ctx: ReaderCtx,
entry: PartitionEntry,
reason: 'graceful' | 'lost',
runtime: ReaderRuntime
): void {
entry.session.stop()
entry.state = 'stopped'
entry.requestedCommitOffset = undefined
ctx.sessionIndex.delete(entry.partitionSessionId)
runtime.emit({
type: 'reader.partition.stopped',
partitionId: entry.partitionId,
reason,
session: entry.session,
})
}
let stopPartitionSession = function stopPartitionSession(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.stop_partition' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let key = ctx.sessionIndex.get(event.partitionSessionId)
let entry = key !== undefined ? ctx.partitions.get(key) : undefined
Iif (!entry || key === undefined || entry.state === 'stopped') {
return []
}
// The stop request carries the server's committed high-water mark — resolve every
// pending commit it already covers before deciding whether anything is left to
// wait for (otherwise a graceful stop waits on commits the server already holds).
advanceCommitted(entry, event.committedOffset, runtime)
let confirmedOverride =
entry.requestedCommitOffset !== undefined &&
entry.partitionCommittedOffset >= entry.requestedCommitOffset
if (confirmedOverride) {
entry.requestedCommitOffset = undefined
}
for (let waiterId of drainCommits(entry, entry.partitionCommittedOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
if (!event.graceful) {
// Immediate: give up the partition. Hold pending commits for reconcile if the
// partition comes back; bound the wait with a gc timer (rebalanced-away case).
markStopped(ctx, entry, 'lost', runtime)
let effects: ReaderEffect[] = [
{
type: 'reader.effect.timer.clear',
which: 'partition_commit_status',
partitionKey: key,
},
// A pending graceful escalated to force — its fallback timer is obsolete.
{
type: 'reader.effect.timer.clear',
which: 'partition_graceful_timeout',
partitionKey: key,
},
]
if (entry.pendingCommits.length > 0) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionKey: key,
})
}
return effects
}
// Graceful (soft stop): the partition stays deliverable and committable while the
// app finishes. The stop response goes out only after BOTH the async
// onPartitionSessionStop hook completes (stop_ready) and the pending commits
// drain — bounded by the per-partition timeout, since the server waits for the
// response with no timeout of its own.
entry.state = 'stopping-graceful'
entry.stopReady = false
let effects: ReaderEffect[] = [
{
type: 'reader.effect.partition.stop_hook',
partitionSessionId: event.partitionSessionId,
partitionKey: key,
grantId: entry.grantId,
committedOffset: entry.partitionCommittedOffset,
},
{
type: 'reader.effect.timer.schedule',
which: 'partition_graceful_timeout',
partitionKey: key,
},
]
if (confirmedOverride) {
effects.push({
type: 'reader.effect.timer.clear',
which: 'partition_commit_status',
partitionKey: key,
})
}
return effects
}
// The async onPartitionSessionStart hook finished — answer the server, unless the
// grant was superseded while the hook ran: grantId pins the exact grant (session ids
// collide across reconnects — they restart at 1 per stream), and ackPending dedupes
// a double completion. Only now is the session fully live: the reassign gc is
// released and the pending commits are sent, strictly AFTER the start response and
// exactly once (recordCommit buffers while ackPending).
let ackPartitionStart = function ackPartitionStart(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.partition.start_ready' }>
): ReaderEffect[] {
let entry = ctx.partitions.get(event.partitionKey)
if (!entry || entry.grantId !== event.grantId || !entry.ackPending) {
return []
}
if (
entry.state !== 'active' ||
ctx.sessionIndex.get(entry.partitionSessionId) !== event.partitionKey
) {
return []
}
entry.ackPending = false
// Start overrides have no commit ack. Suppress overlapping sends on this grant,
// but retain pending ranges until a server watermark confirms the commit.
if (event.commitOffset !== undefined && event.commitOffset > entry.partitionCommittedOffset) {
entry.requestedCommitOffset = event.commitOffset
}
let effects: ReaderEffect[] = [
{
type: 'reader.effect.send.start_response',
partitionSessionId: event.partitionSessionId,
...(event.readOffset !== undefined && { readOffset: event.readOffset }),
...(event.commitOffset !== undefined && { commitOffset: event.commitOffset }),
},
{
type: 'reader.effect.timer.clear',
which: 'partition_reassign_gc',
partitionKey: event.partitionKey,
},
]
// The single send of everything buffered for this partition (reconciled pendings
// from before the reconnect and commits issued during the hook window alike) —
// each pending replays its exact remaining ranges, never a gap-covering span.
for (let pending of entry.pendingCommits) {
let ranges = compactCoverage(
pending.wireRanges,
entry.requestedCommitOffset ?? entry.partitionCommittedOffset
)
if (ranges.length > 0) {
effects.push(commitEffect(entry.partitionSessionId, ranges))
}
}
if (entry.requestedCommitOffset !== undefined) {
effects.push({
type: 'reader.effect.send.partition_status',
partitionSessionId: entry.partitionSessionId,
})
}
return effects
}
// The async onPartitionSessionStop hook finished — the graceful stop may be
// acknowledged once the pending commits drain too. Guards mirror ackPartitionStart:
// grantId pins the grant, state pins the phase (a force stop or reconnect while the
// hook ran makes this a no-op).
let ackPartitionStop = function ackPartitionStop(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.partition.stop_ready' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let entry = ctx.partitions.get(event.partitionKey)
if (!entry || entry.grantId !== event.grantId || entry.state !== 'stopping-graceful') {
return []
}
entry.stopReady = true
if (entry.pendingCommits.length > 0 || entry.requestedCommitOffset !== undefined) {
return []
}
let sessionId = entry.partitionSessionId
let live = ctx.sessionIndex.get(sessionId) === event.partitionKey
markStopped(ctx, entry, 'graceful', runtime)
let effects: ReaderEffect[] = [
{
type: 'reader.effect.timer.clear',
which: 'partition_commit_status',
partitionKey: event.partitionKey,
},
{
type: 'reader.effect.timer.clear',
which: 'partition_graceful_timeout',
partitionKey: event.partitionKey,
},
]
// Answer only over the stream that asked (session ids restart per stream).
if (live) {
effects.push(stopResponseEffect(sessionId))
}
return effects
}
// A graceful stop that never became answerable in time (hook hung or commits never
// drained): acknowledge the stop anyway (the server otherwise waits forever and the
// partition is never handed off) and fall back to the force-stop bookkeeping —
// pending commits stay for a possible reconcile, bounded by the reassign gc.
let forceStopStalledGraceful = function forceStopStalledGraceful(
ctx: ReaderCtx,
key: string,
runtime: ReaderRuntime
): ReaderEffect[] {
let entry = ctx.partitions.get(key)
if (!entry || entry.state !== 'stopping-graceful') {
return []
}
let effects: ReaderEffect[] = [
{ type: 'reader.effect.timer.clear', which: 'partition_commit_status', partitionKey: key },
]
entry.session.stop()
entry.state = 'stopped'
entry.requestedCommitOffset = undefined
runtime.emit({
type: 'reader.partition.stopped',
partitionId: entry.partitionId,
reason: 'graceful',
session: entry.session,
})
// Answer the server only when the session id still belongs to THIS stream (same
// predicate as recordCommit): server-assigned ids restart at 1 per stream, so a
// stale id from before a reconnect likely names a different, freshly granted
// partition — releasing it is session-fatal (BAD_REQUEST) and deleting its index
// mapping would silently drop that partition's reads. A stale stop needs no
// answer at all: the server re-requests it on the new session if still relevant.
if (ctx.sessionIndex.get(entry.partitionSessionId) === key) {
ctx.sessionIndex.delete(entry.partitionSessionId)
effects.push(stopResponseEffect(entry.partitionSessionId))
}
if (entry.pendingCommits.length > 0) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionKey: key,
})
}
return effects
}
let commitOffsetResponse = function commitOffsetResponse(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.commit_response' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let effects: ReaderEffect[] = []
for (let committed of event.committed) {
let key = ctx.sessionIndex.get(committed.partitionSessionId)
let entry = key !== undefined ? ctx.partitions.get(key) : undefined
if (!entry || key === undefined) {
continue
}
advanceCommitted(entry, committed.committedOffset, runtime)
if (
entry.requestedCommitOffset !== undefined &&
entry.partitionCommittedOffset >= entry.requestedCommitOffset
) {
entry.requestedCommitOffset = undefined
effects.push({
type: 'reader.effect.timer.clear',
which: 'partition_commit_status',
partitionKey: key,
})
}
for (let waiterId of drainCommits(entry, entry.partitionCommittedOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
// A graceful stop that was waiting on these commits can now be acknowledged —
// once the stop hook has completed too — for the exact session the server
// asked to stop (== the acked one).
if (
entry.state === 'stopping-graceful' &&
entry.stopReady &&
entry.requestedCommitOffset === undefined &&
entry.pendingCommits.length === 0
) {
markStopped(ctx, entry, 'graceful', runtime)
effects.push(stopResponseEffect(committed.partitionSessionId), {
type: 'reader.effect.timer.clear',
which: 'partition_graceful_timeout',
partitionKey: key,
})
}
}
return effects
}
let endPartitionSession = function endPartitionSession(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.end_partition' }>,
runtime: ReaderRuntime
): void {
let key = ctx.sessionIndex.get(event.partitionSessionId)
let entry = key !== undefined ? ctx.partitions.get(key) : undefined
Iif (!entry) {
return
}
// Partition fully read (autopartitioning split/merge). No response; keep the entry
// AND its session index so pending/late commits still reach the wire — the session
// stays open for commits until the server stops it. The child/adjacent partition
// ids are recorded on the session for autoscaling-aware consumers.
entry.session.end(event.childPartitionIds, event.adjacentPartitionIds)
entry.state = 'ended'
runtime.emit({
type: 'reader.partition.stopped',
partitionId: entry.partitionId,
reason: 'ended',
session: entry.session,
})
}
// Dispatch a typed server-stream event to its handler. Returns effects to append.
let applyStreamEvent = function applyStreamEvent(
ctx: ReaderCtx,
event: ReaderStreamEvent,
runtime: ReaderRuntime
): ReaderEffect[] {
switch (event.type) {
case 'reader.stream.read_response':
return readResponse(ctx, event, runtime)
case 'reader.stream.start_partition':
return startPartitionSession(ctx, event, runtime)
case 'reader.stream.stop_partition':
return stopPartitionSession(ctx, event, runtime)
case 'reader.stream.commit_response':
return commitOffsetResponse(ctx, event, runtime)
case 'reader.stream.partition_status': {
let effects = commitOffsetResponse(
ctx,
{
type: 'reader.stream.commit_response',
committed: [
{
partitionSessionId: event.partitionSessionId,
committedOffset: event.committedOffset,
},
],
},
runtime
)
let key = ctx.sessionIndex.get(event.partitionSessionId)
let entry = key !== undefined ? ctx.partitions.get(key) : undefined
if (entry?.requestedCommitOffset !== undefined && key !== undefined) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_commit_status',
partitionKey: key,
})
}
return effects
}
case 'reader.stream.end_partition':
endPartitionSession(ctx, event, runtime)
return []
}
}
// Buffer a commit at the partition level; send it now if we have a live stream.
let recordCommit = function recordCommit(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.commit' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let entry = ctx.partitions.get(event.partitionKey)
Iif (!entry) {
// The partition is gone (stopped + gc'd) — nothing can acknowledge it.
runtime.emit({
type: 'reader.commit.rejected',
waiterId: event.waiterId,
reason: new Error(`No active partition ${event.partitionKey} to commit`),
})
return []
}
let requestedRanges = mergeRanges(event.ranges)
let targetOffset = requestedRanges.at(-1)?.end ?? entry.partitionCommittedOffset
if (targetOffset <= entry.partitionCommittedOffset) {
runtime.emit({ type: 'reader.commit.resolved', waiterId: event.waiterId })
return []
}
// Clamp to server truth and subtract already-claimed coverage: a duplicate or
// overlapping range on the wire is session-fatal. A fully claimed commit still
// remains pending until the server confirms its target watermark.
let wireRanges = subtractCovered(
requestedRanges,
entry.claimedRanges,
entry.partitionCommittedOffset
)
// A commit can lose the race with the partition stop: the facade dispatches while
// the session is still live, but a commit_response or stop request queued ahead of
// it completes the stop first. Uncovered offsets of a stopped partition can never
// be acknowledged on this stream (the server ignores commits after the stop), and
// the messages will be redelivered wherever the partition lands — reject now
// instead of parking the waiter on the reassign gc.
if (entry.state === 'stopped') {
runtime.emit({
type: 'reader.commit.rejected',
waiterId: event.waiterId,
reason: new Error(
`Cannot commit offsets for a stopped or expired partition session (partition ${event.partitionKey})`
),
})
return []
}
entry.pendingCommits.push({ targetOffset, wireRanges, waiterId: event.waiterId })
entry.claimedRanges = mergeRanges([...entry.claimedRanges, ...wireRanges])
// Send only in `ready` and only over a session granted by the CURRENT stream:
// toReady clears sessionIndex and only start_partition repopulates it, so a commit
// landing in the init→start_partition window after a reconnect buffers here and
// rides the reconcile — sending immediately would use the previous stream's
// session id.
// 'stopping-graceful' still sends: the protocol accepts commits until the client
// responds to the stop request, and the commit_response drain relies on them.
// 'ended' sends too — end_partition is informational (partition fully read); the
// session stays open for commits until the server stops it.
// While a grant's start handshake is outstanding (ackPending) the commit is
// buffered too: the ack performs the single send — sending here AND there would
// put the same range on the wire twice, which is session-fatal.
let sessionLive =
runtime.state === 'ready' &&
ctx.sessionIndex.get(entry.partitionSessionId) === event.partitionKey
let committable =
entry.state === 'active' || entry.state === 'stopping-graceful' || entry.state === 'ended'
if (sessionLive && committable && !entry.ackPending && wireRanges.length > 0) {
let ranges = compactCoverage(
wireRanges,
entry.requestedCommitOffset ?? entry.partitionCommittedOffset
)
return ranges.length > 0 ? [commitEffect(entry.partitionSessionId, ranges)] : []
}
return []
}
// Replenish exactly the server-accounted size of responses that have fully passed
// through read(). This keeps the stream's credit window stable even when one response
// exceeds the initial grant.
let releaseBytes = function releaseBytes(ctx: ReaderCtx, bytes: bigint): ReaderEffect[] {
ctx.inFlightBytes -= bytes
if (ctx.inFlightBytes < 0n) {
ctx.inFlightBytes = 0n
}
return bytes > 0n ? [readRequestEffect(bytes)] : []
}
// ── Terminal / transitions ──────────────────────────────────────────────────────
let terminate = function terminate(
ctx: ReaderCtx,
state: 'closed' | 'errored',
reason: unknown,
runtime: ReaderRuntime
): TransitionResult<ReaderState, ReaderEffect> {
if (state === 'errored') {
ctx.lastError = reason
runtime.emit({ type: 'reader.error', error: reason })
}
// Reject every outstanding commit exactly once — the facade settles the waiterId.
for (let entry of ctx.partitions.values()) {
for (let pending of entry.pendingCommits) {
runtime.emit({ type: 'reader.commit.rejected', waiterId: pending.waiterId, reason })
}
entry.pendingCommits = []
}
runtime.emit({ type: 'reader.closed', reason })
releaseState(ctx)
return {
state,
// Terminal: the runtime seals itself after the finalize effect runs, so the
// buffered lifecycle outputs (reader.closed / rejects) are delivered first.
final: { reason },
effects: [
{ type: 'reader.effect.transport.close' },
{ type: 'reader.effect.timer.clear', which: 'start_timeout' },
{ type: 'reader.effect.timer.clear', which: 'retry_backoff' },
{ type: 'reader.effect.timer.clear', which: 'recovery_window' },
{ type: 'reader.effect.timer.clear', which: 'update_token' },
{ type: 'reader.effect.timer.clear', which: 'graceful_timeout' },
{ type: 'reader.effect.finalize', reason },
],
}
}
let releaseState = function releaseState(ctx: ReaderCtx): void {
ctx.partitions.clear()
ctx.sessionIndex.clear()
ctx.inFlightBytes = 0n
}
// Enter `ready` on a successful init. Unlike the writer there is no seqNo recovery:
// the server re-sends start_partition per partition, where reconcile happens.
let toReady = function toReady(
ctx: ReaderCtx,
sessionId: string,
runtime: ReaderRuntime
): TransitionResult<ReaderState, ReaderEffect> {
ctx.sessionId = sessionId
ctx.hasEverConnected = true
ctx.attempts = 0
// Ephemeral session ids from the previous stream are dead; buffered ReadResponses
// on it are gone. Reset flow-control and re-issue the full initial credit (the new
// stream grants a fresh maxBufferBytes budget, so old pending credit is moot).
ctx.sessionIndex.clear()
ctx.inFlightBytes = 0n
runtime.emit({ type: 'reader.session', sessionId })
let effects: ReaderEffect[] = [
...clearConnectTimersEffects,
{ type: 'reader.effect.timer.schedule', which: 'update_token' },
readRequestEffect(ctx.limits.maxBufferBytes),
]
// Bound the wait for every partition holding pending commits: if the server does
// not re-grant it on this stream (rebalanced to another reader), the gc rejects
// the waiters instead of leaving them pending forever. start_partition clears the
// timer when the partition does come back.
for (let [key, entry] of ctx.partitions) {
if (entry.pendingCommits.length > 0) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionKey: key,
})
}
}
return { state: 'ready', effects }
}
let toReconnecting = function toReconnecting(
ctx: ReaderCtx,
error: unknown,
runtime: ReaderRuntime
): TransitionResult<ReaderState, ReaderEffect> {
// sessionIndex means "session ids granted by the CURRENT stream" — with the stream
// gone there are none. Clearing here (not only in toReady) keeps the guards in
// recordCommit / forceStopStalledGraceful honest while connecting: nothing may be
// sent under an id the next stream never granted.
ctx.sessionIndex.clear()
if (error !== undefined) {
ctx.lastError = error
}
runtime.emit({
type: 'reader.reconnecting',
attempt: ctx.attempts,
...(error !== undefined && { error }),
})
let effects: ReaderEffect[] = [
{ type: 'reader.effect.timer.clear', which: 'start_timeout' },
{ type: 'reader.effect.timer.clear', which: 'update_token' },
{ type: 'reader.effect.timer.schedule', which: 'retry_backoff' },
]
for (let key of ctx.partitions.keys()) {
effects.push({
type: 'reader.effect.timer.clear',
which: 'partition_commit_status',
partitionKey: key,
})
}
// Arm the terminal deadline only when recovery is bounded. Unbounded (Infinity)
// means reconnect forever — the transition owns that policy so the emitted effects
// reflect it (model-testable), instead of the runtime silently dropping the timer.
if (Number.isFinite(ctx.recoveryWindowMs)) {
effects.push({ type: 'reader.effect.timer.schedule', which: 'recovery_window' })
}
return { state: 'reconnecting', effects }
}
let toClosing = function toClosing(
ctx: ReaderCtx,
runtime: ReaderRuntime
): TransitionResult<ReaderState, ReaderEffect> {
if (!hasPendingWork(ctx)) {
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
}
return {
state: 'closing',
effects: [
// Closing may be entered while connecting/reconnecting — cancel those timers
// so a stale start_timeout/retry_backoff/recovery_window cannot fire against
// the closing drain.
...clearConnectTimersEffects,
{ type: 'reader.effect.timer.clear', which: 'update_token' },
{ type: 'reader.effect.timer.schedule', which: 'graceful_timeout' },
],
}
}
let hasPendingWork = function hasPendingWork(ctx: ReaderCtx): boolean {
for (let entry of ctx.partitions.values()) {
if (
entry.pendingCommits.length > 0 ||
entry.requestedCommitOffset !== undefined ||
entry.state === 'stopping-graceful'
) {
return true
}
}
return false
}
// A partition that was stopped (rebalanced away) and never came back: reject its
// still-pending commits so the caller is not left hanging (at-least-once redelivery
// covers correctness; the messages go to the partition's new owner).
let gcPartition = function gcPartition(ctx: ReaderCtx, key: string, runtime: ReaderRuntime): void {
let entry = ctx.partitions.get(key)
Iif (!entry) {
return
}
// A granted-and-acked session is live — the ack cleared this timer, so a firing
// against it is a stale race; ignore it.
Iif (ctx.sessionIndex.get(entry.partitionSessionId) === key && !entry.ackPending) {
return
}
// Two reapable cases: the partition was never re-granted on this stream
// (rebalanced to another reader), or it was granted but the start handshake never
// completed (a hung onPartitionSessionStart hook) — either way the waiters would
// hang until terminal close. Reject them; keep an un-acked granted entry so a
// late start_ready can still answer the server (it just has nothing to re-send).
for (let pending of entry.pendingCommits) {
runtime.emit({
type: 'reader.commit.rejected',
waiterId: pending.waiterId,
reason: new Error(`Partition ${key} reassigned before commit was acknowledged`),
})
}
entry.pendingCommits = []
entry.claimedRanges = []
if (ctx.sessionIndex.get(entry.partitionSessionId) !== key) {
ctx.partitions.delete(key)
}
}
// ── Transition ──────────────────────────────────────────────────────────────────
// Deliberately-ignored (state, event) pairs route through here so an unhandled
// event shows up in debug logs instead of vanishing.
let ignored = function ignored(state: ReaderState, event: ReaderEvent): void {
dbg.log('ignoring %s in state %s', event.type, state)
}
export let readerTransition = function readerTransition(
ctx: ReaderCtx,
event: ReaderEvent,
runtime: ReaderRuntime
): TransitionResult<ReaderState, ReaderEffect> | void {
let state = runtime.state
// Global: hard destroy from any non-terminal state.
if (state !== 'closed' && state !== 'errored' && event.type === 'reader.destroy') {
return terminate(ctx, 'closed', event.reason ?? new Error('Reader destroyed'), runtime)
}
if (
(state === 'ready' || state === 'closing') &&
event.type === 'reader.timer.partition_commit_status'
) {
let entry = ctx.partitions.get(event.partitionKey)
if (
entry?.requestedCommitOffset !== undefined &&
ctx.sessionIndex.get(entry.partitionSessionId) === event.partitionKey
) {
return {
effects: [
{
type: 'reader.effect.send.partition_status',
partitionSessionId: entry.partitionSessionId,
},
],
}
}
return
}
switch (state) {
case 'idle': {
switch (event.type) {
case 'reader.start':
return {
state: 'connecting',
effects: [
{ type: 'reader.effect.transport.connect' },
{ type: 'reader.effect.timer.schedule', which: 'start_timeout' },
],
}
case 'reader.close':
return terminate(
ctx,
'closed',
new Error('Reader closed before start'),
runtime
)
default:
return ignored(state, event)
}
}
case 'connecting':
case 'reconnecting': {
switch (event.type) {
case 'reader.stream.init_response':
return toReady(ctx, event.sessionId, runtime)
case 'reader.commit':
// Buffered for re-send on the next start_partition (recordCommit never
// sends outside `ready`).
return { effects: recordCommit(ctx, event, runtime) }
case 'reader.stream.disconnected':
case 'reader.timer.start_timeout': {
let error =
event.type === 'reader.stream.disconnected' ? event.error : undefined
if (
event.type === 'reader.stream.disconnected' &&
!isRetryableReaderError(error, ctx.retryOnSchemeError)
) {
return terminate(ctx, 'errored', error, runtime)
}
return toReconnecting(ctx, error, runtime)
}
case 'reader.timer.retry_backoff':
Eif (state === 'reconnecting') {
ctx.attempts += 1
return {
state: 'connecting',
effects: [
{ type: 'reader.effect.transport.connect' },
{ type: 'reader.effect.timer.schedule', which: 'start_timeout' },
],
}
}
// A stale backoff firing while already connecting has nothing to do.
return ignored(state, event)
case 'reader.timer.recovery_window':
return terminate(
ctx,
'errored',
ctx.lastError ?? new Error('Reader recovery window expired'),
runtime
)
// Partition timers armed in `ready` survive the disconnect edge — handle
// them here so a stalled graceful stop / lost partition is still bounded.
// forceStopStalledGraceful suppresses the stop_response for sessions the
// current stream did not grant; the server re-requests the stop on the next
// session if the rebalance is still in progress.
case 'reader.timer.partition_graceful_timeout': {
let effects = forceStopStalledGraceful(ctx, event.partitionKey, runtime)
return { effects }
}
case 'reader.timer.partition_reassign_gc':
gcPartition(ctx, event.partitionKey, runtime)
return
case 'reader.close':
return toClosing(ctx, runtime)
default:
return ignored(state, event)
}
}
case 'ready': {
switch (event.type) {
case 'reader.stream.read_response':
case 'reader.stream.start_partition':
case 'reader.stream.stop_partition':
case 'reader.stream.commit_response':
case 'reader.stream.partition_status':
case 'reader.stream.end_partition': {
let effects = applyStreamEvent(ctx, event, runtime)
return { effects }
}
case 'reader.commit':
return { effects: recordCommit(ctx, event, runtime) }
case 'reader.partition.start_ready': {
let effects = ackPartitionStart(ctx, event)
return { effects }
}
case 'reader.partition.stop_ready': {
let effects = ackPartitionStop(ctx, event, runtime)
return { effects }
}
case 'reader.read_release': {
let effects = releaseBytes(ctx, event.bytes)
return { effects }
}
case 'reader.timer.update_token':
return { effects: [{ type: 'reader.effect.send.update_token' }] }
case 'reader.timer.partition_reassign_gc':
gcPartition(ctx, event.partitionKey, runtime)
return
// Fallback for a graceful stop whose hook or commits never completed: the
// server waits for the stop response indefinitely, so the client must not.
case 'reader.timer.partition_graceful_timeout': {
let effects = forceStopStalledGraceful(ctx, event.partitionKey, runtime)
return { effects }
}
case 'reader.stream.disconnected':
if (!isRetryableReaderError(event.error, ctx.retryOnSchemeError)) {
return terminate(ctx, 'errored', event.error, runtime)
}
return toReconnecting(ctx, event.error, runtime)
case 'reader.close':
return toClosing(ctx, runtime)
default:
return ignored(state, event)
}
}
case 'closing': {
switch (event.type) {
case 'reader.stream.read_response':
case 'reader.stream.start_partition':
case 'reader.stream.stop_partition':
case 'reader.stream.commit_response':
case 'reader.stream.partition_status':
case 'reader.stream.end_partition': {
let effects = applyStreamEvent(ctx, event, runtime)
if (!hasPendingWork(ctx)) {
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
}
return { effects }
}
// start_partition is honored during the closing drain, so its async hook
// completion must be answered here too.
case 'reader.partition.start_ready': {
let effects = ackPartitionStart(ctx, event)
return { effects }
}
// The graceful-stop drain continues during close(); a completed stop hook
// may finish the drain and with it the whole close.
case 'reader.partition.stop_ready': {
let effects = ackPartitionStop(ctx, event, runtime)
if (!hasPendingWork(ctx)) {
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
}
return { effects }
}
// Per-partition fallback (armed in ready) — force-stop that partition and
// keep draining; only the global close deadline finalizes the reader.
case 'reader.timer.partition_graceful_timeout': {
let effects = forceStopStalledGraceful(ctx, event.partitionKey, runtime)
if (!hasPendingWork(ctx)) {
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
}
return { effects }
}
case 'reader.timer.graceful_timeout':
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
case 'reader.stream.disconnected':
// A drop mid-close abandons any un-acked commits — finalize.
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
case 'reader.timer.partition_reassign_gc':
gcPartition(ctx, event.partitionKey, runtime)
if (!hasPendingWork(ctx)) {
return terminate(ctx, 'closed', new Error('Reader closed'), runtime)
}
return
default:
return ignored(state, event)
}
}
case 'closed':
case 'errored':
return ignored(state, event)
}
}
|