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 | 13x 13x 11529x 13x 711x 347x 364x 9x 355x 13x 13x 11732x 13x 297x 13x 272x 13x 358x 358x 17303x 4x 17299x 17299x 356x 16943x 16943x 16942x 1x 358x 358x 356x 358x 13x 1791x 1791x 1791x 326x 164x 162x 1791x 1791x 13x 801x 801x 948x 17786x 801x 13x 802x 802x 803x 803x 803x 2x 801x 801x 801x 802x 802x 802x 13x 1227x 1227x 1227x 1227x 890x 890x 337x 337x 337x 337x 337x 337x 337x 1x 337x 337x 337x 1227x 1227x 13x 1227x 1227x 1227x 1227x 1227x 1227x 1227x 4x 1227x 78x 2x 1227x 1227x 1227x 13x 381x 381x 381x 381x 13x 409x 409x 409x 409x 4x 4x 409x 3x 409x 122x 122x 20x 102x 287x 244x 244x 43x 43x 13x 1194x 1194x 2x 1192x 1x 1191x 1191x 3x 1x 3x 3x 3x 1191x 1194x 77x 1191x 13x 21x 21x 32x 11x 21x 1x 20x 20x 20x 20x 13x 13x 20x 20x 21x 13x 153x 153x 153x 153x 153x 1x 152x 152x 152x 152x 152x 156x 152x 15x 15x 153x 13x 3x 3x 3x 3x 3x 3x 13x 2594x 802x 1227x 409x 153x 3x 3x 13x 358x 358x 358x 358x 2x 2x 356x 356x 358x 358x 220x 136x 13x 13x 433x 433x 23x 433x 433x 433x 9x 9x 9x 424x 13x 11477x 4x 4x 11477x 807x 127x 807x 11477x 11477x 11477x 13x 11477x 11477x 11477x 11477x 13x 11723x 11723x 11723x 11723x 11723x 11723x 11723x 11723x 11723x 369x 84x 11723x 13x 992x 992x 359x 992x 992x 992x 1x 992x 13x 428x 379x 49x 13x 465x 411x 73x 392x 13x 49x 49x 14x 35x 35x 40x 35x 35x 34x 13x 139x 13x 41262x 41262x 11046x 30216x 11529x 11529x 12989x 11723x 79x 294x 294x 1x 293x 512x 512x 512x 5x 5x 13x 13x 247x 116x 5594x 2565x 2565x 279x 1184x 1184x 433x 433x 205x 31x 31x 13x 13x 702x 3x 699x 181x 1x 103x 29x 29x 8x 21x 10x 10x 3x 3x 3x 27x 8x 5x 5x 5x 21x 1x | 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
// partitionId, never the ephemeral partitionSessionId (proto: "unique inside one
// RPC call"). 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'
// A commit awaiting the server's high-water-mark ack. `waiterId` maps to a Promise
// in the facade; the transition only records the offset range.
export type PendingCommit = {
startOffset: bigint
endOffset: bigint
waiterId: number
}
// Everything the reader tracks about one partition, keyed by the stable partitionId.
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: { start: bigint; end: bigint }
partitionCommittedOffset: bigint
// Gap-fill anchor: the start of the next commit range. Survives reconnect and is
// never rewound below a start_partition committed_offset (see startPartitionSession).
nextCommitStartOffset: bigint
// lifecycle
state: 'active' | 'stopping-graceful' | 'stopped' | 'ended'
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
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 STABLE partitionId
partitions: Map<bigint, PartitionEntry>
// ephemeral partitionSessionId -> partitionId, rebuilt each stream
sessionIndex: Map<bigint, bigint>
// monotonic grant counter — source of PartitionEntry.grantId
grantSeq: number
// byte flow-control
inFlightBytes: bigint
pendingReadRequestBytes: 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 partitionId and key as `<name>:<partitionId>`.
export type PartitionTimerName = 'partition_graceful_timeout' | 'partition_reassign_gc'
export type TimerName = GlobalTimerName | PartitionTimerName
// Timer control payload — partitionId is required exactly when the name is partition_*.
export type TimerRef =
| { which: GlobalTimerName }
| { which: PartitionTimerName; partitionId: bigint }
// 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.start_partition'
partitionSessionId: bigint
partitionId: bigint
path: string
committedOffset: bigint
partitionOffsets: { start: bigint; end: bigint }
}
| {
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 }
export type ReaderEvent =
// user (dispatched by the facade)
| { type: 'reader.start' }
| { type: 'reader.commit'; partitionId: bigint; offsets: bigint[]; 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
partitionId: bigint
grantId: number
readOffset?: bigint
commitOffset?: bigint
}
// 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'; partitionId: bigint }
| { type: 'reader.timer.partition_reassign_gc'; partitionId: bigint }
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: { start: bigint; end: bigint }[]
}
| { type: 'reader.effect.send.stop_response'; 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
partitionId: bigint
grantId: number
committedOffset: bigint
partitionOffsets: { start: bigint; end: 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'
}
| { type: 'reader.partition.committed'; partitionId: bigint; committedOffset: bigint }
| { 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,
pendingReadRequestBytes: 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: { start: bigint; end: bigint }[]
): ReaderEffect {
return { type: 'reader.effect.send.commit', partitionSessionId, ranges }
}
let stopResponseEffect = function stopResponseEffect(partitionSessionId: bigint): ReaderEffect {
return { type: 'reader.effect.send.stop_response', partitionSessionId }
}
// Group a partition's sorted offsets into disjoint [start, end) ranges, filling the
// gap between the last commit point (nextCommitStartOffset — covers retention-deleted
// offsets) and the first message. Advances nextCommitStartOffset (never rewinds it).
let buildCommitRanges = function buildCommitRanges(
entry: PartitionEntry,
offsets: bigint[]
): { start: bigint; end: bigint }[] {
let ranges: { start: bigint; end: bigint }[] = []
for (let offset of offsets) {
// Already covered by the commit high-water mark (re-committed message, or a
// redelivery below the anchor). Skipping is load-bearing, not just tidy: a
// zero-width or inverted OffsetsRange is session-fatal — the server answers
// CloseSession(BAD_REQUEST "double committing is forbidden").
if (offset + 1n <= entry.nextCommitStartOffset) {
continue
}
let last = ranges[ranges.length - 1]
if (last === undefined) {
// First range starts at the gap-fill anchor so retention gaps are covered.
ranges.push({ start: entry.nextCommitStartOffset, end: offset + 1n })
} else Iif (offset + 1n <= last.end) {
// Duplicate / already covered — skip (facade validates order, defensive here).
continue
} else if (offset === last.end) {
last.end = offset + 1n
} else {
ranges.push({ start: offset, end: offset + 1n })
}
}
let lastRange = ranges[ranges.length - 1]
if (lastRange !== undefined) {
entry.nextCommitStartOffset = lastRange.end
}
return ranges
}
// 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.endOffset <= committedOffset) {
resolved.push(pending.waiterId)
} else {
kept.push(pending)
}
}
entry.pendingCommits = kept
return resolved
}
// Decode one ReadResponse partition into messages. 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(partitionData: PartitionReadData): ReaderMessage[] {
let messages: ReaderMessage[] = []
for (let batch of partitionData.batches) {
for (let md of batch.messageData) {
messages.push({
offset: md.offset,
seqNo: md.seqNo,
data: md.data,
uncompressedSize: md.uncompressedSize,
producer: batch.producerId,
codec: batch.codec,
...(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 partitionId = ctx.sessionIndex.get(partitionData.partitionSessionId)
let entry = partitionId !== undefined ? ctx.partitions.get(partitionId) : undefined
// Drop data for an unknown, superseded (stale session id after a reassign), or
// no-longer-active partition.
if (
!entry ||
entry.partitionSessionId !== partitionData.partitionSessionId ||
entry.state === 'stopped' ||
entry.state === 'ended'
) {
continue
}
let messages = collectMessages(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 and the gap-fill anchor survive.
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 entry = ctx.partitions.get(partitionId)
ctx.grantSeq += 1
if (entry === undefined) {
entry = {
partitionId,
path,
partitionSessionId,
session,
grantId: ctx.grantSeq,
ackPending: true,
partitionOffsets,
partitionCommittedOffset: committedOffset,
nextCommitStartOffset: committedOffset,
state: 'active',
pendingCommits: [],
}
ctx.partitions.set(partitionId, 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 + the gap-fill anchor.
ctx.sessionIndex.delete(entry.partitionSessionId)
entry.session = session
entry.partitionSessionId = partitionSessionId
entry.path = path
entry.partitionOffsets = partitionOffsets
entry.partitionCommittedOffset = committedOffset
// Never rewind the gap-fill anchor below server truth.
if (entry.nextCommitStartOffset < committedOffset) {
entry.nextCommitStartOffset = committedOffset
}
entry.state = 'active'
entry.grantId = ctx.grantSeq
entry.ackPending = true
}
ctx.sessionIndex.set(partitionSessionId, partitionId)
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 session = new TopicPartitionSession(partitionSessionId, partitionId, path)
session.partitionOffsets = partitionOffsets
session.partitionCommittedOffset = committedOffset
let entry = upsertPartitionEntry(ctx, event, session)
let effects: ReaderEffect[] = [
// 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', partitionId },
]
// COMMIT RECONCILE, half one: resolve pending already covered by committed_offset
// and narrow the remainder to [committed_offset, end). 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).
for (let waiterId of drainCommits(entry, committedOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
for (let pending of entry.pendingCommits) {
if (pending.startOffset < committedOffset) {
pending.startOffset = committedOffset
}
}
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,
partitionId,
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'
ctx.sessionIndex.delete(entry.partitionSessionId)
runtime.emit({ type: 'reader.partition.stopped', partitionId: entry.partitionId, reason })
}
let stopPartitionSession = function stopPartitionSession(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.stop_partition' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let partitionId = ctx.sessionIndex.get(event.partitionSessionId)
let entry = partitionId !== undefined ? ctx.partitions.get(partitionId) : undefined
Iif (!entry || 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).
if (event.committedOffset > entry.partitionCommittedOffset) {
entry.partitionCommittedOffset = event.committedOffset
entry.session.partitionCommittedOffset = event.committedOffset
}
for (let waiterId of drainCommits(entry, event.committedOffset)) {
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)
if (entry.pendingCommits.length > 0) {
return [
{
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionId: entry.partitionId,
},
]
}
return []
}
// Graceful: process pending commits, then respond. If already drained, respond now.
if (entry.pendingCommits.length === 0) {
markStopped(ctx, entry, 'graceful', runtime)
return [stopResponseEffect(event.partitionSessionId)]
}
entry.state = 'stopping-graceful'
// The response is sent once pending commits drain (commit_response) or the
// per-partition graceful timeout fires — the server waits for the response with no
// timeout of its own, so the fallback must exist and must not collide across
// concurrently stopping partitions.
return [
{
type: 'reader.effect.timer.schedule',
which: 'partition_graceful_timeout',
partitionId: entry.partitionId,
},
]
}
// 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' }>,
runtime: ReaderRuntime
): ReaderEffect[] {
let entry = ctx.partitions.get(event.partitionId)
if (!entry || entry.grantId !== event.grantId || !entry.ackPending) {
return []
}
if (
entry.state !== 'active' ||
ctx.sessionIndex.get(entry.partitionSessionId) !== entry.partitionId
) {
return []
}
entry.ackPending = false
// A commitOffset override moves the server's committed mark: reconcile pending
// commits against it exactly like startPartitionSession does with the server's
// committed offset — resolve fully covered waiters, narrow the rest, advance the
// gap-fill anchor. Re-sending a range below the override is session-fatal.
if (event.commitOffset !== undefined) {
for (let waiterId of drainCommits(entry, event.commitOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
for (let pending of entry.pendingCommits) {
if (pending.startOffset < event.commitOffset) {
pending.startOffset = event.commitOffset
}
}
Eif (event.commitOffset > entry.nextCommitStartOffset) {
entry.nextCommitStartOffset = 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',
partitionId: entry.partitionId,
},
]
// The single send of everything buffered for this partition (reconciled pendings
// from before the reconnect and commits issued during the hook window alike).
for (let pending of entry.pendingCommits) {
effects.push(
commitEffect(entry.partitionSessionId, [
{ start: pending.startOffset, end: pending.endOffset },
])
)
}
return effects
}
// A graceful stop whose pending commits never drained in time: 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,
partitionId: bigint,
runtime: ReaderRuntime
): ReaderEffect[] {
let effects: ReaderEffect[] = []
for (let entry of ctx.partitions.values()) {
if (entry.state !== 'stopping-graceful') {
continue
}
if (entry.partitionId !== partitionId) {
continue
}
entry.session.stop()
entry.state = 'stopped'
runtime.emit({
type: 'reader.partition.stopped',
partitionId: entry.partitionId,
reason: 'graceful',
})
// 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) === entry.partitionId) {
ctx.sessionIndex.delete(entry.partitionSessionId)
effects.push(stopResponseEffect(entry.partitionSessionId))
}
Eif (entry.pendingCommits.length > 0) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionId: entry.partitionId,
})
}
}
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 partitionId = ctx.sessionIndex.get(committed.partitionSessionId)
let entry = partitionId !== undefined ? ctx.partitions.get(partitionId) : undefined
if (!entry) {
continue
}
Eif (committed.committedOffset > entry.partitionCommittedOffset) {
entry.partitionCommittedOffset = committed.committedOffset
}
entry.session.partitionCommittedOffset = committed.committedOffset
runtime.emit({
type: 'reader.partition.committed',
partitionId: entry.partitionId,
committedOffset: committed.committedOffset,
})
for (let waiterId of drainCommits(entry, committed.committedOffset)) {
runtime.emit({ type: 'reader.commit.resolved', waiterId })
}
// A graceful stop that was waiting on these commits can now be acknowledged —
// for the exact session the server asked to stop (== the acked one).
if (entry.state === 'stopping-graceful' && entry.pendingCommits.length === 0) {
markStopped(ctx, entry, 'graceful', runtime)
effects.push(stopResponseEffect(committed.partitionSessionId), {
type: 'reader.effect.timer.clear',
which: 'partition_graceful_timeout',
partitionId: entry.partitionId,
})
}
}
return effects
}
let endPartitionSession = function endPartitionSession(
ctx: ReaderCtx,
event: Extract<ReaderEvent, { type: 'reader.stream.end_partition' }>,
runtime: ReaderRuntime
): void {
let partitionId = ctx.sessionIndex.get(event.partitionSessionId)
let entry = partitionId !== undefined ? ctx.partitions.get(partitionId) : undefined
Iif (!entry) {
return
}
// Partition fully read (split/merge). No response; keep the entry so pending
// commits still reconcile against a future committed_offset or terminal shutdown.
entry.session.end()
entry.state = 'ended'
runtime.emit({
type: 'reader.partition.stopped',
partitionId: entry.partitionId,
reason: 'ended',
})
}
// 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.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.partitionId)
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.partitionId} to commit`),
})
return []
}
let ranges = buildCommitRanges(entry, event.offsets)
if (ranges.length === 0) {
runtime.emit({ type: 'reader.commit.resolved', waiterId: event.waiterId })
return []
}
entry.pendingCommits.push({
startOffset: ranges[0]!.start,
endOffset: ranges[ranges.length - 1]!.end,
waiterId: event.waiterId,
})
// 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) === entry.partitionId
let committable =
entry.state === 'active' || entry.state === 'stopping-graceful' || entry.state === 'ended'
if (sessionLive && committable && !entry.ackPending) {
return [commitEffect(entry.partitionSessionId, ranges)]
}
return []
}
// Re-grant server read credit only once at least 1/5 of the buffer budget is
// pending — batches the ReadRequests instead of sending one per consumed response.
let CREDIT_REGRANT_DIVISOR = 5n
// Consumer released `bytes`; replenish the server credit past a threshold.
let releaseBytes = function releaseBytes(ctx: ReaderCtx, bytes: bigint): ReaderEffect[] {
ctx.inFlightBytes -= bytes
if (ctx.inFlightBytes < 0n) {
ctx.inFlightBytes = 0n
}
ctx.pendingReadRequestBytes += bytes
// Ceil division so a budget smaller than the divisor still yields a non-zero
// threshold.
let threshold =
(ctx.limits.maxBufferBytes + CREDIT_REGRANT_DIVISOR - 1n) / CREDIT_REGRANT_DIVISOR
if (ctx.pendingReadRequestBytes >= threshold) {
let credit = ctx.pendingReadRequestBytes
ctx.pendingReadRequestBytes = 0n
return [readRequestEffect(credit)]
}
return []
}
// ── 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
ctx.pendingReadRequestBytes = 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
ctx.pendingReadRequestBytes = 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 entry of ctx.partitions.values()) {
if (entry.pendingCommits.length > 0) {
effects.push({
type: 'reader.effect.timer.schedule',
which: 'partition_reassign_gc',
partitionId: entry.partitionId,
})
}
}
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' },
]
// 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.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,
partitionId: bigint,
runtime: ReaderRuntime
): void {
let entry = ctx.partitions.get(partitionId)
if (!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) === entry.partitionId && !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 ${partitionId} reassigned before commit was acknowledged`),
})
}
entry.pendingCommits = []
if (ctx.sessionIndex.get(entry.partitionSessionId) !== entry.partitionId) {
ctx.partitions.delete(partitionId)
}
}
// ── 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)
}
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.partitionId, runtime)
return { effects }
}
case 'reader.timer.partition_reassign_gc':
gcPartition(ctx, event.partitionId, 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.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, 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.partitionId, runtime)
return
// Fallback for a graceful stop whose commits never drained: the server waits
// for the stop response indefinitely, so the client must not.
case 'reader.timer.partition_graceful_timeout': {
let effects = forceStopStalledGraceful(ctx, event.partitionId, 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.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, 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.partitionId, runtime)
Iif (!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.partitionId, runtime)
Eif (!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)
}
}
|