1import { PgProtocol } from "@effect/sql-pg"
2import { assert, describe, it } from "@effect/vitest"
3import * as Result from "effect/Result"
4import { backend, bytes, frontend } from "./fixtures/goldens.ts"
5
6const hex = (value: Uint8Array): string => Array.from(value, (byte) => byte.toString(16).padStart(2, "0")).join("")
7
8const success = <A, E>(result: Result.Result<A, E>): A => {
9 assert.isTrue(Result.isSuccess(result))
10 return (result as Result.Success<A, E>).success
11}
12
13const parseOne = (golden: string): PgProtocol.BackendMessage => {
14 const messages = PgProtocol.makeParser().push(bytes(golden))
15 assert.strictEqual(messages.length, 1)
16 return messages[0]
17}
18
19const assertThrowsTagged = (tag: string, run: () => unknown, message?: string): void => {
20 try {
21 run()
22 } catch (error) {
23 assert.strictEqual((error as { readonly _tag?: string })._tag, tag)
24 if (message !== undefined) assert.strictEqual((error as Error).message, message)
25 return
26 }
27 assert.fail(`Expected ${tag}`)
28}
29
30const assertParseFailure = (golden: string): void =>
31 assertThrowsTagged("PgProtocolParseError", () => PgProtocol.makeParser().push(bytes(golden)))
32
33describe("PgProtocol", () => {
34 describe("special messages", () => {
35 it("encodes an SSLRequest", () => {
36 assert.strictEqual(hex(PgProtocol.encodeSslRequest()), frontend.sslRequest)
37 })
38
39 it("decodes the SSLRequest response byte", () => {
40 assert.strictEqual(success(PgProtocol.decodeSslResponse(0x53)), "S")
41 assert.strictEqual(success(PgProtocol.decodeSslResponse(0x4e)), "N")
42 const invalid = PgProtocol.decodeSslResponse(0x41)
43 assert.isTrue(Result.isFailure(invalid))
44 if (Result.isFailure(invalid)) assert.strictEqual(invalid.failure._tag, "PgProtocolParseError")
45 })
46
47 it("encodes a StartupMessage", () => {
48 const encoded = PgProtocol.encodeStartupMessage({
49 user: "effect",
50 database: "effect",
51 application_name: "effect-pg-codec"
52 })
53 assert.strictEqual(hex(encoded), frontend.startupMessage)
54 })
55
56 it("keeps an explicit client_encoding", () => {
57 const encoded = PgProtocol.encodeStartupMessage({ user: "effect", client_encoding: "UTF8" })
58 assert.strictEqual(
59 hex(encoded),
60 "0000002a00030000757365720065666665637400636c69656e745f656e636f64696e6700555446380000"
61 )
62 })
63
64 it("encodes a CancelRequest", () => {
65 const encoded = PgProtocol.encodeCancelRequest({ pid: 63, secret: 166060928 })
66 assert.strictEqual(hex(encoded), frontend.cancelRequest)
67 })
68 })
69
70 describe("frontend messages", () => {
71 it("encodes Parse", () => {
72 assert.strictEqual(
73 hex(success(PgProtocol.encodeParse({ name: "s1", query: "SELECT $1", parameterTypes: [23] }))),
74 frontend.parse
75 )
76 })
77
78 it("encodes Bind with binary parameter and result formats", () => {
79 const encoded = success(PgProtocol.encodeBind({
80 portal: "p1",
81 statement: "s1",
82 parameters: [bytes("00000001"), null]
83 }))
84 assert.strictEqual(hex(encoded), frontend.bind)
85 })
86
87 it("encodes strings past the ASCII fast path", () => {
88 const encoder = new TextEncoder()
89 for (
90 const query of [
91 "SELECT $1",
92 `SELECT ${"c".repeat(200)} FROM t WHERE id = $1`,
93 "SELECT 'héllo ☃' , '👋🏽' FROM t WHERE id = $1",
94 "é",
95 ""
96 ]
97 ) {
98 const encoded = success(PgProtocol.encodeParse({ name: "s1", query, parameterTypes: [23] }))
99 const body = encoder.encode(query)
100 assert.deepStrictEqual(encoded.subarray(8, 8 + body.length), body)
101 assert.strictEqual(encoded[8 + body.length], 0)
102 assert.strictEqual(encoded.length, 8 + body.length + 1 + 2 + 4)
103 }
104 })
105
106 it("encodes Bind through a value sink", () => {
107 const encodeBind = PgProtocol.makeBindEncoder<Uint8Array | null>((sink, value) => {
108 if (value === null) sink.sqlNull()
109 else sink.raw(value)
110 return Result.void
111 })
112 assert.strictEqual(
113 hex(success(encodeBind({ portal: "p1", statement: "s1", parameters: [bytes("00000001"), null] }))),
114 frontend.bind
115 )
116 })
117
118 it("frames each sink-written parameter with the bytes it wrote", () => {
119 const encodeBind = PgProtocol.makeBindEncoder<number>((sink, value) => {
120 sink.int16(value)
121 sink.utf8("é")
122 sink.float64(value)
123 return Result.void
124 })
125 const encoded = success(encodeBind({ portal: "", statement: "", parameters: [1, 2] }))
126 // 1 type byte, 4 length, 2 empty cStrings, 3 int16 of format codes and
127 // count, then each parameter as its int32 length and 12 bytes of body
128 const view = new DataView(encoded.buffer, encoded.byteOffset, encoded.byteLength)
129 assert.strictEqual(view.getInt32(13), 12)
130 assert.strictEqual(view.getInt16(17), 1)
131 assert.strictEqual(view.getFloat64(21), 1)
132 assert.strictEqual(view.getInt32(29), 12)
133 assert.strictEqual(view.getFloat64(37), 2)
134 })
135
136 it("keeps sink-written parameters intact when the pool grows mid-message", () => {
137 const encodeBind = PgProtocol.makeBindEncoder<string>((sink, value) => {
138 sink.utf8(value)
139 return Result.void
140 })
141 const long = "x".repeat(16 * 1024)
142 const encoded = success(encodeBind({ portal: "", statement: "", parameters: ["ab", long, "cd"] }))
143 assert.deepStrictEqual(
144 encoded,
145 success(PgProtocol.encodeBind({
146 portal: "",
147 statement: "",
148 parameters: ["ab", long, "cd"].map((value) => new TextEncoder().encode(value))
149 }))
150 )
151 })
152
153 it("encodes Execute", () => {
154 assert.strictEqual(hex(PgProtocol.encodeExecute({ portal: "p1", maxRows: 5 })), frontend.execute)
155 })
156
157 it("encodes Describe and Close", () => {
158 assert.strictEqual(
159 hex(PgProtocol.encodeDescribe({ target: "statement", name: "s1" })),
160 frontend.describeStatement
161 )
162 assert.strictEqual(hex(PgProtocol.encodeClose({ target: "portal", name: "p1" })), frontend.closePortal)
163 })
164
165 it("encodes the empty control messages", () => {
166 assert.strictEqual(hex(PgProtocol.encodeSync()), frontend.sync)
167 assert.strictEqual(hex(PgProtocol.encodeFlush()), frontend.flush)
168 assert.strictEqual(hex(PgProtocol.encodeTerminate()), frontend.terminate)
169 })
170
171 it("encodes the password messages", () => {
172 assert.strictEqual(
173 hex(PgProtocol.encodePasswordMessage({ password: "md5abc" })),
174 frontend.passwordMessage
175 )
176 assert.strictEqual(
177 hex(PgProtocol.encodeSASLInitialResponse({
178 mechanism: "SCRAM-SHA-256",
179 initialResponse: bytes("6e2c2c")
180 })),
181 frontend.saslInitialResponse
182 )
183 assert.strictEqual(
184 hex(PgProtocol.encodeSASLInitialResponse({ mechanism: "SCRAM-SHA-256", initialResponse: null })),
185 frontend.saslInitialResponseEmpty
186 )
187 assert.strictEqual(hex(PgProtocol.encodeSASLResponse({ data: bytes("010203") })), frontend.saslResponse)
188 })
189
190 it("dispatches encode over the tagged union", () => {
191 const cases: ReadonlyArray<readonly [PgProtocol.FrontendMessage, Uint8Array]> = [
192 [
193 { _tag: "Parse", name: "s1", query: "SELECT $1", parameterTypes: [23] },
194 success(PgProtocol.encodeParse({ name: "s1", query: "SELECT $1", parameterTypes: [23] }))
195 ],
196 [
197 { _tag: "Bind", portal: "p1", statement: "s1", parameters: [bytes("00000001"), null] },
198 success(PgProtocol.encodeBind({ portal: "p1", statement: "s1", parameters: [bytes("00000001"), null] }))
199 ],
200 [{ _tag: "Execute", portal: "p1", maxRows: 5 }, PgProtocol.encodeExecute({ portal: "p1", maxRows: 5 })],
201 [
202 { _tag: "Describe", target: "statement", name: "s1" },
203 PgProtocol.encodeDescribe({ target: "statement", name: "s1" })
204 ],
205 [
206 { _tag: "Close", target: "portal", name: "p1" },
207 PgProtocol.encodeClose({ target: "portal", name: "p1" })
208 ],
209 [{ _tag: "Sync" }, PgProtocol.encodeSync()],
210 [{ _tag: "Flush" }, PgProtocol.encodeFlush()],
211 [{ _tag: "Terminate" }, PgProtocol.encodeTerminate()],
212 [
213 { _tag: "PasswordMessage", password: "md5abc" },
214 PgProtocol.encodePasswordMessage({ password: "md5abc" })
215 ],
216 [
217 { _tag: "SASLInitialResponse", mechanism: "SCRAM-SHA-256", initialResponse: bytes("6e2c2c") },
218 PgProtocol.encodeSASLInitialResponse({ mechanism: "SCRAM-SHA-256", initialResponse: bytes("6e2c2c") })
219 ],
220 [{ _tag: "SASLResponse", data: bytes("010203") }, PgProtocol.encodeSASLResponse({ data: bytes("010203") })]
221 ]
222 for (const [message, expected] of cases) {
223 assert.deepStrictEqual(success(PgProtocol.encode(message)), expected)
224 }
225 })
226
227 it("writes the actual typed frame length into every message", () => {
228 const messages = [
229 success(PgProtocol.encodeParse({ name: "s1", query: "SELECT $1", parameterTypes: [23] })),
230 success(PgProtocol.encodeBind({ portal: "p1", statement: "s1", parameters: [bytes("00000001"), null] })),
231 PgProtocol.encodeExecute({ portal: "p1", maxRows: 5 }),
232 PgProtocol.encodeDescribe({ target: "statement", name: "s1" }),
233 PgProtocol.encodeClose({ target: "portal", name: "p1" }),
234 PgProtocol.encodeSync(),
235 PgProtocol.encodeFlush(),
236 PgProtocol.encodeTerminate(),
237 PgProtocol.encodePasswordMessage({ password: "md5abc" }),
238 PgProtocol.encodeSASLInitialResponse({ mechanism: "SCRAM-SHA-256", initialResponse: bytes("6e2c2c") }),
239 PgProtocol.encodeSASLResponse({ data: bytes("010203") })
240 ]
241 for (const message of messages) {
242 assert.strictEqual(
243 new DataView(message.buffer, message.byteOffset, message.byteLength).getInt32(1),
244 message.length - 1
245 )
246 }
247 })
248
249 it("accepts the maximum signed parameter count and rejects the next value", () => {
250 const maxOids = new Array<number>(0x7fff).fill(23)
251 const maxParameters = new Array<Uint8Array | null>(0x7fff).fill(null)
252 const parse = success(PgProtocol.encodeParse({ name: "", query: "", parameterTypes: maxOids }))
253 const bind = success(PgProtocol.encodeBind({ portal: "", statement: "", parameters: maxParameters }))
254 assert.strictEqual(new DataView(parse.buffer, parse.byteOffset, parse.byteLength).getInt16(7), 0x7fff)
255 assert.strictEqual(new DataView(bind.buffer, bind.byteOffset, bind.byteLength).getInt16(11), 0x7fff)
256
257 const tooManyOids = new Array<number>(0x8000).fill(23)
258 const tooManyParameters = new Array<Uint8Array | null>(0x8000).fill(null)
259 const parseFailure = PgProtocol.encodeParse({ name: "", query: "", parameterTypes: tooManyOids })
260 const bindFailure = PgProtocol.encodeBind({ portal: "", statement: "", parameters: tooManyParameters })
261 assert.isTrue(Result.isFailure(parseFailure))
262 assert.isTrue(Result.isFailure(bindFailure))
263 const encodeBind = PgProtocol.makeBindEncoder<Uint8Array | null>((sink, value) => {
264 if (value === null) sink.sqlNull()
265 else sink.raw(value)
266 return Result.void
267 })
268 const fusedFailure = encodeBind({ portal: "", statement: "", parameters: tooManyParameters })
269 assert.isTrue(Result.isFailure(fusedFailure))
270 })
271 })
272
273 describe("authentication messages", () => {
274 it("decodes AuthenticationOk", () => {
275 assert.deepStrictEqual(parseOne(backend.authenticationOk), { _tag: "AuthenticationOk" })
276 })
277
278 it("decodes AuthenticationCleartextPassword", () => {
279 assert.deepStrictEqual(parseOne(backend.authenticationCleartextPassword), {
280 _tag: "AuthenticationCleartextPassword"
281 })
282 })
283
284 it("decodes AuthenticationMD5Password", () => {
285 assert.deepStrictEqual(parseOne(backend.authenticationMD5Password), {
286 _tag: "AuthenticationMD5Password",
287 salt: bytes("70e7d45e")
288 })
289 })
290
291 it("decodes AuthenticationSASL", () => {
292 assert.deepStrictEqual(parseOne(backend.authenticationSASL), {
293 _tag: "AuthenticationSASL",
294 mechanisms: ["SCRAM-SHA-256"]
295 })
296 })
297
298 it("decodes the SASL continuation payloads as opaque bytes", () => {
299 const cont = parseOne(backend.authenticationSASLContinue)
300 assert.strictEqual(cont._tag, "AuthenticationSASLContinue")
301 assert.strictEqual(
302 new TextDecoder().decode((cont as PgProtocol.AuthenticationSASLContinue).data).slice(0, 2),
303 "r="
304 )
305 const final = parseOne(backend.authenticationSASLFinal)
306 assert.strictEqual(final._tag, "AuthenticationSASLFinal")
307 })
308
309 it("reports unsupported authentication methods", () => {
310 // AuthenticationGSS, method 7
311 assert.deepStrictEqual(parseOne("520000000800000007"), {
312 _tag: "AuthenticationUnsupported",
313 method: 7,
314 payload: new Uint8Array(0)
315 })
316 })
317
318 it("rejects truncated authentication fields", () => {
319 for (
320 const frame of [
321 "5200000007000000", // truncated authentication method
322 "520000000b00000005010203", // three-byte MD5 salt
323 "520000000e0000000a534352414d00" // SASL mechanism list without its empty terminator
324 ]
325 ) {
326 assertParseFailure(frame)
327 }
328 })
329 })
330
331 describe("backend messages", () => {
332 it("decodes ParameterStatus", () => {
333 assert.deepStrictEqual(parseOne(backend.parameterStatus), {
334 _tag: "ParameterStatus",
335 name: "in_hot_standby",
336 value: "off"
337 })
338 })
339
340 it("decodes BackendKeyData", () => {
341 assert.deepStrictEqual(parseOne(backend.backendKeyData), {
342 _tag: "BackendKeyData",
343 pid: 63,
344 secret: 166060928
345 })
346 })
347
348 it("decodes ReadyForQuery", () => {
349 assert.deepStrictEqual(parseOne(backend.readyForQuery), { _tag: "ReadyForQuery", status: "I" })
350 })
351
352 // ParameterStatus is two C-strings and nothing else, so it is the cheapest
353 // way to drive the string decoder across the length its fast path covers.
354 const parameterStatus = (name: string, value: string): Uint8Array => {
355 const encoder = new TextEncoder()
356 const encode = (text: string) => encoder.encode(text)
357 return frame(encode(name), encode(value))
358 }
359
360 const frame = (name: Uint8Array, value: Uint8Array): Uint8Array => {
361 const length = 4 + name.length + 1 + value.length + 1
362 const message = new Uint8Array(1 + length)
363 message[0] = 0x53
364 new DataView(message.buffer).setInt32(1, length)
365 message.set(name, 5)
366 message.set(value, 6 + name.length)
367 return message
368 }
369
370 it("decodes parameter names of every length across the string fast path", () => {
371 for (let length = 0; length <= 16; length++) {
372 const name = "abcdefgh".repeat(2).slice(0, length)
373 assert.deepStrictEqual(PgProtocol.makeParser().push(parameterStatus(name, "v")), [
374 { _tag: "ParameterStatus", name, value: "v" }
375 ])
376 }
377 })
378
379 it("decodes a multi-byte character at every position of the string fast path", () => {
380 for (let prefix = 0; prefix <= 16; prefix++) {
381 const name = "a".repeat(prefix) + "é☃"
382 assert.deepStrictEqual(PgProtocol.makeParser().push(parameterStatus(name, "v")), [
383 { _tag: "ParameterStatus", name, value: "v" }
384 ])
385 }
386 })
387
388 it("decodes a parameter that really contains the replacement character", () => {
389 const name = "status\ufffdmessage and some padding"
390 assert.deepStrictEqual(
391 PgProtocol.makeParser().push(parameterStatus(name, "v")),
392 [{ _tag: "ParameterStatus", name, value: "v" }]
393 )
394 })
395
396 it("rejects a stray high byte at every position of the string fast path", () => {
397 for (let length = 1; length <= 16; length++) {
398 for (let at = 0; at < length; at++) {
399 const name = new Uint8Array(length).fill(0x61)
400 name[at] = 0xc3
401 assertThrowsTagged(
402 "PgProtocolParseError",
403 () => PgProtocol.makeParser().push(frame(name, new Uint8Array([0x76])))
404 )
405 }
406 }
407 })
408
409 it("decodes RowDescription", () => {
410 assert.deepStrictEqual(parseOne(backend.rowDescription), {
411 _tag: "RowDescription",
412 fields: [
413 {
414 name: "a",
415 tableOid: 0,
416 columnAttributeNumber: 0,
417 dataTypeOid: 23,
418 dataTypeSize: 4,
419 typeModifier: -1,
420 format: 1
421 },
422 {
423 name: "b",
424 tableOid: 0,
425 columnAttributeNumber: 0,
426 dataTypeOid: 25,
427 dataTypeSize: -1,
428 typeModifier: -1,
429 format: 1
430 }
431 ]
432 })
433 })
434
435 it("decodes DataRow as raw bytes, with null for SQL NULL", () => {
436 assert.deepStrictEqual(parseOne(backend.dataRowTwoColumns), {
437 _tag: "DataRow",
438 values: [bytes("00000001"), bytes("78")]
439 })
440 assert.deepStrictEqual(parseOne(backend.dataRowWithNull), {
441 _tag: "DataRow",
442 values: [bytes("0000000000000007"), null]
443 })
444 })
445
446 it("decodes ParameterDescription", () => {
447 assert.deepStrictEqual(parseOne(backend.parameterDescription), {
448 _tag: "ParameterDescription",
449 parameterTypes: [20]
450 })
451 })
452
453 it("decodes CommandComplete", () => {
454 assert.deepStrictEqual(parseOne(backend.commandComplete), {
455 _tag: "CommandComplete",
456 commandTag: "SELECT 1"
457 })
458 })
459
460 it("decodes the empty acknowledgements", () => {
461 assert.deepStrictEqual(parseOne(backend.parseComplete), { _tag: "ParseComplete" })
462 assert.deepStrictEqual(parseOne(backend.bindComplete), { _tag: "BindComplete" })
463 assert.deepStrictEqual(parseOne(backend.closeComplete), { _tag: "CloseComplete" })
464 assert.deepStrictEqual(parseOne(backend.portalSuspended), { _tag: "PortalSuspended" })
465 assert.deepStrictEqual(parseOne(backend.emptyQueryResponse), { _tag: "EmptyQueryResponse" })
466 assert.deepStrictEqual(parseOne(backend.noData), { _tag: "NoData" })
467 })
468
469 it("decodes ErrorResponse fields by name", () => {
470 assert.deepStrictEqual(parseOne(backend.errorResponse), {
471 _tag: "ErrorResponse",
472 fields: {
473 severity: "ERROR",
474 severityUnlocalized: "ERROR",
475 code: "22012",
476 message: "division by zero",
477 file: "int.c",
478 line: "870",
479 routine: "int4div"
480 }
481 })
482 })
483
484 it("decodes NoticeResponse", () => {
485 const notice = parseOne(backend.noticeResponse)
486 assert.strictEqual(notice._tag, "NoticeResponse")
487 assert.strictEqual((notice as PgProtocol.NoticeResponse).fields.code, "00000")
488 assert.strictEqual((notice as PgProtocol.NoticeResponse).fields.severity, "NOTICE")
489 })
490
491 it("decodes NotificationResponse", () => {
492 assert.deepStrictEqual(parseOne(backend.notificationResponse), {
493 _tag: "NotificationResponse",
494 pid: 63,
495 channel: "effect_channel",
496 payload: "payload text"
497 })
498 })
499
500 it("decodes NegotiateProtocolVersion", () => {
501 // minor version 0, one unrecognised option "_pq_.foo"
502 assert.deepStrictEqual(parseOne("760000001500000000000000015f70715f2e666f6f00"), {
503 _tag: "NegotiateProtocolVersion",
504 minorVersion: 0,
505 unrecognizedOptions: ["_pq_.foo"]
506 })
507 })
508
509 it("decodes the COPY messages", () => {
510 assert.deepStrictEqual(parseOne("470000000b00000200000001"), {
511 _tag: "CopyInResponse",
512 format: 0,
513 columnFormats: [0, 1]
514 })
515 assert.deepStrictEqual(parseOne("48000000090100010000"), {
516 _tag: "CopyOutResponse",
517 format: 1,
518 columnFormats: [0]
519 })
520 assert.deepStrictEqual(parseOne("57000000090100010000"), {
521 _tag: "CopyBothResponse",
522 format: 1,
523 columnFormats: [0]
524 })
525 assert.deepStrictEqual(parseOne("6400000008010203ff"), {
526 _tag: "CopyData",
527 data: bytes("010203ff")
528 })
529 assert.deepStrictEqual(parseOne("6300000004"), { _tag: "CopyDone" })
530 assert.deepStrictEqual(parseOne("6400000004"), { _tag: "CopyData", data: new Uint8Array(0) })
531 })
532
533 it("reports an unknown type byte instead of failing", () => {
534 assert.deepStrictEqual(parseOne("5f00000008deadbeef"), {
535 _tag: "Unknown",
536 type: 0x5f,
537 payload: bytes("deadbeef")
538 })
539 assert.deepStrictEqual(parseOne("5f00000004"), {
540 _tag: "Unknown",
541 type: 0x5f,
542 payload: new Uint8Array(0)
543 })
544 })
545 })
546
547 describe("incremental parsing", () => {
548 it("buffers a message split across two chunks", () => {
549 const parser = PgProtocol.makeParser()
550 const frame = bytes(backend.dataRowTwoColumns)
551 assert.deepStrictEqual(parser.push(frame.subarray(0, 7)), [])
552 const messages = parser.push(frame.subarray(7))
553 assert.strictEqual(messages.length, 1)
554 assert.strictEqual(messages[0]._tag, "DataRow")
555 })
556
557 it("pushEach hands over a message before it reads the next", () => {
558 const parser = PgProtocol.makeParser<unknown>()
559 const seen: Array<PgProtocol.BackendMessage<unknown>> = []
560 parser.pushEach(bytes(backend.rowDescription + backend.dataRowTwoColumns), (message) => {
561 // A row's columns are only known from the description in front of it,
562 // so installing the reader here has to reach the rows behind it.
563 if (message._tag === "RowDescription") {
564 parser.readField = (bytes, offset, size) => (size < 0 ? null : hex(bytes.subarray(offset, offset + size)))
565 }
566 seen.push(message)
567 })
568 assert.strictEqual(seen.length, 2)
569 assert.strictEqual(seen[0]._tag, "RowDescription")
570 assert.deepStrictEqual(seen[1], { _tag: "DataRow", values: ["00000001", "78"] })
571 })
572
573 it("pushEach keeps the messages it handed over before a failure", () => {
574 const parser = PgProtocol.makeParser()
575 const seen: Array<PgProtocol.BackendMessage> = []
576 assertThrowsTagged(
577 "PgProtocolParseError",
578 () => parser.pushEach(bytes(backend.parseComplete + "4480000000"), (message) => seen.push(message))
579 )
580 assert.deepStrictEqual(seen, [{ _tag: "ParseComplete" }])
581 })
582
583 it("returns both messages when one chunk holds two", () => {
584 const parser = PgProtocol.makeParser()
585 const chunk = bytes(backend.parseComplete + backend.bindComplete)
586 assert.deepStrictEqual(parser.push(chunk), [{ _tag: "ParseComplete" }, { _tag: "BindComplete" }])
587 })
588
589 it("returns every message when one chunk holds many", () => {
590 const parser = PgProtocol.makeParser()
591 const messages = parser.push(bytes(backend.parseComplete.repeat(2_000)))
592 assert.strictEqual(messages.length, 2_000)
593 assert.strictEqual(messages.every((message) => message._tag === "ParseComplete"), true)
594 })
595
596 it("keeps the trailing partial message across pushes", () => {
597 const parser = PgProtocol.makeParser()
598 const chunk = bytes(backend.parseComplete + backend.readyForQuery)
599 assert.deepStrictEqual(parser.push(chunk.subarray(0, 8)), [{ _tag: "ParseComplete" }])
600 assert.deepStrictEqual(parser.push(chunk.subarray(8)), [{ _tag: "ReadyForQuery", status: "I" }])
601 })
602
603 it("handles a byte-at-a-time stream", () => {
604 const parser = PgProtocol.makeParser()
605 const frame = bytes(backend.rowDescription)
606 const messages: Array<PgProtocol.BackendMessage> = []
607 for (let i = 0; i < frame.length; i++) {
608 messages.push(...parser.push(frame.subarray(i, i + 1)))
609 }
610 assert.strictEqual(messages.length, 1)
611 assert.strictEqual(messages[0]._tag, "RowDescription")
612 })
613
614 it("grows its buffer for a large fragmented DataRow", () => {
615 const field = new Uint8Array(40 * 1024)
616 for (let index = 0; index < field.length; index++) field[index] = index % 251
617 const frame = new Uint8Array(11 + field.length)
618 const view = new DataView(frame.buffer)
619 frame[0] = 0x44
620 view.setInt32(1, frame.length - 1)
621 view.setInt16(5, 1)
622 view.setInt32(7, field.length)
623 frame.set(field, 11)
624
625 const parser = PgProtocol.makeParser()
626 const messages: Array<PgProtocol.BackendMessage> = []
627 for (let offset = 0; offset < frame.length; offset += 1024) {
628 messages.push(...parser.push(frame.subarray(offset, offset + 1024)))
629 }
630 assert.strictEqual(messages.length, 1)
631 assert.strictEqual(messages[0]._tag, "DataRow")
632 assert.deepStrictEqual((messages[0] as PgProtocol.DataRow).values, [field])
633 })
634
635 it("rejects a length prefix above maxMessageSize", () => {
636 const parser = PgProtocol.makeParser()
637 assertThrowsTagged("PgProtocolParseError", () => parser.push(bytes("447fffffff0000")))
638 })
639
640 it("honours a custom maxMessageSize", () => {
641 const parser = PgProtocol.makeParser({ maxMessageSize: 8 })
642 assertThrowsTagged("PgProtocolParseError", () => parser.push(bytes(backend.rowDescription)))
643 })
644
645 it("rejects a length prefix below the minimum", () => {
646 const parser = PgProtocol.makeParser()
647 assertThrowsTagged("PgProtocolParseError", () => parser.push(bytes("4400000003ff")))
648 assertThrowsTagged("PgProtocolParseError", () => PgProtocol.makeParser().push(bytes("4480000000")))
649 })
650
651 it("accepts a frame exactly at maxMessageSize", () => {
652 assert.deepStrictEqual(PgProtocol.makeParser({ maxMessageSize: 8 }).push(bytes("5f00000008deadbeef")), [
653 { _tag: "Unknown", type: 0x5f, payload: bytes("deadbeef") }
654 ])
655 })
656
657 it("rejects a truncated payload", () => {
658 // CommandComplete whose command tag is never NUL-terminated
659 assertParseFailure("430000000953454c454354")
660 })
661
662 it("rejects payload bytes beyond a message's declared fields", () => {
663 for (
664 const frame of [
665 "3100000005ff", // ParseComplete with a byte of payload
666 "44000000070000ff" // DataRow declares zero fields but carries a byte
667 ]
668 ) {
669 assertParseFailure(frame)
670 }
671 })
672
673 it("rejects a DataRow field length below the NULL sentinel", () => {
674 assertParseFailure("440000000a0001fffffffe")
675 })
676
677 it("rejects a negative DataRow field count", () => {
678 assertParseFailure("4400000006ffff")
679 })
680
681 it("rejects negative collection counts", () => {
682 for (
683 const frame of [
684 "5400000006ffff", // RowDescription
685 "7400000006ffff", // ParameterDescription
686 "470000000700ffff", // CopyInResponse
687 "480000000700ffff", // CopyOutResponse
688 "570000000700ffff", // CopyBothResponse
689 "760000000c00000000ffffffff" // NegotiateProtocolVersion
690 ]
691 ) {
692 assertParseFailure(frame)
693 }
694 })
695
696 it("rejects a truncated DataRow field length", () => {
697 assertParseFailure("440000000800010000")
698 assertParseFailure("4400000004")
699 })
700
701 it("rejects invalid statuses and UTF-8", () => {
702 assertParseFailure("5a0000000558")
703 assertParseFailure("5300000008ff007800")
704 })
705
706 it("does not read a DataRow field past the end of its own message", () => {
707 // One field that claims ten bytes in a message carrying two, followed by
708 // a complete message the field must not be allowed to reach into
709 assertThrowsTagged(
710 "PgProtocolParseError",
711 () => PgProtocol.makeParser().push(bytes(`440000000c00010000000a0102${backend.parseComplete}`))
712 )
713 })
714
715 it("cannot be reused after a malformed frame", () => {
716 const parser = PgProtocol.makeParser()
717 assertThrowsTagged(
718 "PgProtocolParseError",
719 () => parser.push(bytes(`${backend.parseComplete}430000000953454c454354`))
720 )
721 assertThrowsTagged(
722 "PgProtocolParseError",
723 () => parser.push(bytes(backend.bindComplete)),
724 "Parser cannot be reused after a failure"
725 )
726 })
727
728 it("does not read a string past the end of its own message", () => {
729 // CommandComplete whose tag is not NUL-terminated, followed by a message
730 // whose length prefix contains NUL bytes
731 assertThrowsTagged(
732 "PgProtocolParseError",
733 () => PgProtocol.makeParser().push(bytes(`430000000a53454c454354${backend.parseComplete}`))
734 )
735 })
736 })
737
738 describe("buffer reuse", () => {
739 it("keeps DataRow field views valid across later pushes", () => {
740 const parser = PgProtocol.makeParser()
741 const frame = bytes(backend.dataRowTwoColumns)
742 const first = parser.push(frame)[0] as PgProtocol.DataRow
743 const snapshot = first.values.map((value) => value === null ? null : value.slice())
744 // enough to run past the parser's buffer and force a replacement
745 for (let index = 0; index < 2000; index++) parser.push(frame)
746 assert.deepStrictEqual(first.values, snapshot)
747 })
748
749 it("keeps DataRow field views valid as the buffer pool grows", () => {
750 const parser = PgProtocol.makeParser()
751 const frame = bytes(backend.dataRowTwoColumns)
752 const retained: Array<{ readonly row: PgProtocol.DataRow; readonly snapshot: Array<Uint8Array | null> }> = []
753 // enough pushes to run the pool from its initial size up to its ceiling
754 for (let index = 0; index < 5000; index++) {
755 const row = parser.push(frame)[0] as PgProtocol.DataRow
756 if (index % 500 === 0) {
757 retained.push({ row, snapshot: row.values.map((value) => value === null ? null : value.slice()) })
758 }
759 }
760 for (const { row, snapshot } of retained) {
761 assert.deepStrictEqual(row.values, snapshot)
762 }
763 })
764
765 it("keeps an encoded frame valid after encoding more", () => {
766 const first = success(PgProtocol.encodeBind({
767 portal: "portal",
768 statement: "statement",
769 parameters: [bytes("0102030405")]
770 }))
771 const snapshot = first.slice()
772 for (let index = 0; index < 100; index++) {
773 success(PgProtocol.encodeBind({ portal: "other", statement: "s", parameters: [new Uint8Array(64)] }))
774 }
775 assert.deepStrictEqual(first, snapshot)
776 })
777
778 it("keeps an encoded frame valid after an oversized one", () => {
779 const first = PgProtocol.encodeSync()
780 const snapshot = first.slice()
781 const huge = success(PgProtocol.encodeBind({
782 portal: "",
783 statement: "",
784 parameters: [new Uint8Array(32 * 1024)]
785 }))
786 assert.strictEqual(huge.length, 32 * 1024 + 21)
787 assert.deepStrictEqual(first, snapshot)
788 assert.deepStrictEqual(PgProtocol.encodeSync(), snapshot)
789 })
790 })
791})