qail_pg/protocol/wire/
backend.rs

1//! BackendMessage decoder — server-to-client wire format.
2
3use super::types::*;
4
5/// Maximum backend frame length accepted by the decoder.
6///
7/// Mirrors driver-side guards to keep standalone `BackendMessage::decode`
8/// usage fail-closed against oversized frames.
9pub(crate) const MAX_BACKEND_FRAME_LEN: usize = 64 * 1024 * 1024;
10
11fn decode_utf8(bytes: &[u8], context: &str) -> Result<String, String> {
12    std::str::from_utf8(bytes)
13        .map(str::to_string)
14        .map_err(|e| format!("{} is not valid UTF-8: {}", context, e))
15}
16
17impl BackendMessage {
18    /// Decode a message from wire bytes.
19    pub fn decode(buf: &[u8]) -> Result<(Self, usize), String> {
20        if buf.len() < 5 {
21            return Err("Buffer too short".to_string());
22        }
23
24        let msg_type = buf[0];
25        let len = u32::from_be_bytes([buf[1], buf[2], buf[3], buf[4]]) as usize;
26
27        // PG protocol: length includes itself (4 bytes), so minimum valid length is 4.
28        // Anything less is a malformed message.
29        if len < 4 {
30            return Err(format!("Invalid message length: {} (minimum is 4)", len));
31        }
32        if len > MAX_BACKEND_FRAME_LEN {
33            return Err(format!(
34                "Message too large: {} bytes (max {})",
35                len, MAX_BACKEND_FRAME_LEN
36            ));
37        }
38
39        let frame_len = len
40            .checked_add(1)
41            .ok_or_else(|| "Message length overflow".to_string())?;
42
43        if buf.len() < frame_len {
44            return Err("Incomplete message".to_string());
45        }
46
47        let payload = &buf[5..frame_len];
48
49        let message = match msg_type {
50            b'R' => Self::decode_auth(payload)?,
51            b'S' => Self::decode_parameter_status(payload)?,
52            b'K' => Self::decode_backend_key(payload)?,
53            b'v' => Self::decode_negotiate_protocol_version(payload)?,
54            b'Z' => Self::decode_ready_for_query(payload)?,
55            b'T' => Self::decode_row_description(payload)?,
56            b'D' => Self::decode_data_row(payload)?,
57            b'C' => Self::decode_command_complete(payload)?,
58            b'E' => Self::decode_error_response(payload)?,
59            b'1' => {
60                if !payload.is_empty() {
61                    return Err("ParseComplete must have empty payload".to_string());
62                }
63                BackendMessage::ParseComplete
64            }
65            b'2' => {
66                if !payload.is_empty() {
67                    return Err("BindComplete must have empty payload".to_string());
68                }
69                BackendMessage::BindComplete
70            }
71            b'3' => {
72                if !payload.is_empty() {
73                    return Err("CloseComplete must have empty payload".to_string());
74                }
75                BackendMessage::CloseComplete
76            }
77            b'n' => {
78                if !payload.is_empty() {
79                    return Err("NoData must have empty payload".to_string());
80                }
81                BackendMessage::NoData
82            }
83            b's' => {
84                if !payload.is_empty() {
85                    return Err("PortalSuspended must have empty payload".to_string());
86                }
87                BackendMessage::PortalSuspended
88            }
89            b't' => Self::decode_parameter_description(payload)?,
90            b'G' => Self::decode_copy_in_response(payload)?,
91            b'H' => Self::decode_copy_out_response(payload)?,
92            b'W' => Self::decode_copy_both_response(payload)?,
93            b'd' => BackendMessage::CopyData(payload.to_vec()),
94            b'c' => {
95                if !payload.is_empty() {
96                    return Err("CopyDone must have empty payload".to_string());
97                }
98                BackendMessage::CopyDone
99            }
100            b'A' => Self::decode_notification_response(payload)?,
101            b'I' => {
102                if !payload.is_empty() {
103                    return Err("EmptyQueryResponse must have empty payload".to_string());
104                }
105                BackendMessage::EmptyQueryResponse
106            }
107            b'N' => BackendMessage::NoticeResponse(Self::parse_error_fields(payload)?),
108            _ => return Err(format!("Unknown message type: {}", msg_type as char)),
109        };
110
111        Ok((message, frame_len))
112    }
113
114    fn decode_auth(payload: &[u8]) -> Result<Self, String> {
115        if payload.len() < 4 {
116            return Err("Auth payload too short".to_string());
117        }
118        let auth_type = i32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
119        match auth_type {
120            0 => {
121                if payload.len() != 4 {
122                    return Err(format!(
123                        "AuthenticationOk invalid payload length: {}",
124                        payload.len()
125                    ));
126                }
127                Ok(BackendMessage::AuthenticationOk)
128            }
129            2 => {
130                if payload.len() != 4 {
131                    return Err(format!(
132                        "AuthenticationKerberosV5 invalid payload length: {}",
133                        payload.len()
134                    ));
135                }
136                Ok(BackendMessage::AuthenticationKerberosV5)
137            }
138            3 => {
139                if payload.len() != 4 {
140                    return Err(format!(
141                        "AuthenticationCleartextPassword invalid payload length: {}",
142                        payload.len()
143                    ));
144                }
145                Ok(BackendMessage::AuthenticationCleartextPassword)
146            }
147            5 => {
148                if payload.len() != 8 {
149                    return Err("MD5 auth payload too short (need salt)".to_string());
150                }
151                let mut salt = [0u8; 4];
152                salt.copy_from_slice(&payload[4..8]);
153                Ok(BackendMessage::AuthenticationMD5Password(salt))
154            }
155            6 => {
156                if payload.len() != 4 {
157                    return Err(format!(
158                        "AuthenticationSCMCredential invalid payload length: {}",
159                        payload.len()
160                    ));
161                }
162                Ok(BackendMessage::AuthenticationSCMCredential)
163            }
164            7 => {
165                if payload.len() != 4 {
166                    return Err(format!(
167                        "AuthenticationGSS invalid payload length: {}",
168                        payload.len()
169                    ));
170                }
171                Ok(BackendMessage::AuthenticationGSS)
172            }
173            8 => Ok(BackendMessage::AuthenticationGSSContinue(
174                payload[4..].to_vec(),
175            )),
176            9 => {
177                if payload.len() != 4 {
178                    return Err(format!(
179                        "AuthenticationSSPI invalid payload length: {}",
180                        payload.len()
181                    ));
182                }
183                Ok(BackendMessage::AuthenticationSSPI)
184            }
185            10 => {
186                // SASL - parse mechanism list
187                let mut mechanisms = Vec::new();
188                let mut pos = 4;
189                while pos < payload.len() {
190                    if payload[pos] == 0 {
191                        break; // list terminator
192                    }
193                    let end = payload[pos..]
194                        .iter()
195                        .position(|&b| b == 0)
196                        .map(|p| pos + p)
197                        .ok_or("SASL mechanism list missing null terminator")?;
198                    mechanisms.push(decode_utf8(&payload[pos..end], "SASL mechanism")?);
199                    pos = end + 1;
200                }
201                if pos >= payload.len() {
202                    return Err("SASL mechanism list missing final terminator".to_string());
203                }
204                if pos + 1 != payload.len() {
205                    return Err("SASL mechanism list has trailing bytes".to_string());
206                }
207                if mechanisms.is_empty() {
208                    return Err("SASL mechanism list is empty".to_string());
209                }
210                Ok(BackendMessage::AuthenticationSASL(mechanisms))
211            }
212            11 => {
213                // SASL Continue - server challenge
214                Ok(BackendMessage::AuthenticationSASLContinue(
215                    payload[4..].to_vec(),
216                ))
217            }
218            12 => {
219                // SASL Final - server signature
220                Ok(BackendMessage::AuthenticationSASLFinal(
221                    payload[4..].to_vec(),
222                ))
223            }
224            _ => Err(format!("Unknown auth type: {}", auth_type)),
225        }
226    }
227
228    fn decode_parameter_status(payload: &[u8]) -> Result<Self, String> {
229        let name_end = payload
230            .iter()
231            .position(|&b| b == 0)
232            .ok_or("ParameterStatus missing name terminator")?;
233        if name_end == 0 {
234            return Err("ParameterStatus name is empty".to_string());
235        }
236        let value_start = name_end + 1;
237        if value_start > payload.len() {
238            return Err("ParameterStatus missing value".to_string());
239        }
240        let value_end_rel = payload[value_start..]
241            .iter()
242            .position(|&b| b == 0)
243            .ok_or("ParameterStatus missing value terminator")?;
244        let value_end = value_start + value_end_rel;
245        if value_end + 1 != payload.len() {
246            return Err("ParameterStatus has trailing bytes".to_string());
247        }
248        Ok(BackendMessage::ParameterStatus {
249            name: decode_utf8(&payload[..name_end], "ParameterStatus name")?,
250            value: decode_utf8(&payload[value_start..value_end], "ParameterStatus value")?,
251        })
252    }
253
254    fn decode_backend_key(payload: &[u8]) -> Result<Self, String> {
255        if payload.len() < 8 {
256            return Err("BackendKeyData payload too short".to_string());
257        }
258        let key_len = payload.len() - 4;
259        if !(4..=256).contains(&key_len) {
260            return Err(format!(
261                "BackendKeyData invalid secret key length: {} (expected 4..=256)",
262                key_len
263            ));
264        }
265        let process_id = i32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
266        if process_id <= 0 {
267            return Err(format!("BackendKeyData invalid process id: {process_id}"));
268        }
269        Ok(BackendMessage::BackendKeyData {
270            process_id,
271            secret_key: payload[4..].to_vec(),
272        })
273    }
274
275    fn decode_negotiate_protocol_version(payload: &[u8]) -> Result<Self, String> {
276        if payload.len() < 8 {
277            return Err("NegotiateProtocolVersion payload too short".to_string());
278        }
279
280        let newest_minor_supported =
281            i32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
282        if newest_minor_supported < 0 {
283            return Err("NegotiateProtocolVersion newest_minor_supported is negative".to_string());
284        }
285
286        let unrecognized_count =
287            i32::from_be_bytes([payload[4], payload[5], payload[6], payload[7]]);
288        if unrecognized_count < 0 {
289            return Err(
290                "NegotiateProtocolVersion unrecognized option count is negative".to_string(),
291            );
292        }
293        let unrecognized_count = unrecognized_count as usize;
294        let remaining = payload.len() - 8;
295        // Each option is NUL-terminated, so even an empty option consumes one byte.
296        // This bound keeps malformed counts from forcing huge pre-allocations.
297        if unrecognized_count > remaining {
298            return Err(format!(
299                "NegotiateProtocolVersion unrecognized option count {} exceeds payload capacity {}",
300                unrecognized_count, remaining
301            ));
302        }
303
304        let mut options = Vec::with_capacity(unrecognized_count);
305        let mut pos = 8usize;
306        for _ in 0..unrecognized_count {
307            if pos >= payload.len() {
308                return Err("NegotiateProtocolVersion missing option string terminator".to_string());
309            }
310            let rel_end = payload[pos..]
311                .iter()
312                .position(|&b| b == 0)
313                .ok_or("NegotiateProtocolVersion option missing null terminator")?;
314            let end = pos + rel_end;
315            options.push(decode_utf8(
316                &payload[pos..end],
317                "NegotiateProtocolVersion option",
318            )?);
319            pos = end + 1;
320        }
321
322        if pos != payload.len() {
323            return Err("NegotiateProtocolVersion has trailing bytes".to_string());
324        }
325
326        Ok(BackendMessage::NegotiateProtocolVersion {
327            newest_minor_supported,
328            unrecognized_protocol_options: options,
329        })
330    }
331
332    fn decode_ready_for_query(payload: &[u8]) -> Result<Self, String> {
333        if payload.len() != 1 {
334            return Err("ReadyForQuery payload empty".to_string());
335        }
336        let status = match payload[0] {
337            b'I' => TransactionStatus::Idle,
338            b'T' => TransactionStatus::InBlock,
339            b'E' => TransactionStatus::Failed,
340            _ => return Err("Unknown transaction status".to_string()),
341        };
342        Ok(BackendMessage::ReadyForQuery(status))
343    }
344
345    fn decode_row_description(payload: &[u8]) -> Result<Self, String> {
346        if payload.len() < 2 {
347            return Err("RowDescription payload too short".to_string());
348        }
349
350        let raw_count = i16::from_be_bytes([payload[0], payload[1]]);
351        if raw_count < 0 {
352            return Err(format!("RowDescription invalid field count: {}", raw_count));
353        }
354        let field_count = raw_count as usize;
355        let mut fields = Vec::with_capacity(field_count);
356        let mut pos = 2;
357
358        for _ in 0..field_count {
359            // Field name (null-terminated string)
360            let name_end = payload[pos..]
361                .iter()
362                .position(|&b| b == 0)
363                .ok_or("Missing null terminator in field name")?;
364            let name = decode_utf8(&payload[pos..pos + name_end], "RowDescription field name")?;
365            pos += name_end + 1; // Skip null terminator
366
367            // Ensure we have enough bytes for the fixed fields
368            if pos + 18 > payload.len() {
369                return Err("RowDescription field truncated".to_string());
370            }
371
372            let table_oid = u32::from_be_bytes([
373                payload[pos],
374                payload[pos + 1],
375                payload[pos + 2],
376                payload[pos + 3],
377            ]);
378            pos += 4;
379
380            let column_attr = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
381            pos += 2;
382
383            let type_oid = u32::from_be_bytes([
384                payload[pos],
385                payload[pos + 1],
386                payload[pos + 2],
387                payload[pos + 3],
388            ]);
389            pos += 4;
390
391            let type_size = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
392            pos += 2;
393
394            let type_modifier = i32::from_be_bytes([
395                payload[pos],
396                payload[pos + 1],
397                payload[pos + 2],
398                payload[pos + 3],
399            ]);
400            pos += 4;
401
402            let format = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
403            if !(0..=1).contains(&format) {
404                return Err(format!("RowDescription invalid format code: {}", format));
405            }
406            pos += 2;
407
408            fields.push(FieldDescription {
409                name,
410                table_oid,
411                column_attr,
412                type_oid,
413                type_size,
414                type_modifier,
415                format,
416            });
417        }
418
419        if pos != payload.len() {
420            return Err("RowDescription has trailing bytes".to_string());
421        }
422
423        Ok(BackendMessage::RowDescription(fields))
424    }
425
426    fn decode_data_row(payload: &[u8]) -> Result<Self, String> {
427        if payload.len() < 2 {
428            return Err("DataRow payload too short".to_string());
429        }
430
431        let raw_count = i16::from_be_bytes([payload[0], payload[1]]);
432        if raw_count < 0 {
433            return Err(format!("DataRow invalid column count: {}", raw_count));
434        }
435        let column_count = raw_count as usize;
436        // Sanity check: each column needs at least 4 bytes (length field)
437        if column_count > (payload.len() - 2) / 4 + 1 {
438            return Err(format!(
439                "DataRow claims {} columns but payload is only {} bytes",
440                column_count,
441                payload.len()
442            ));
443        }
444        let mut columns = Vec::with_capacity(column_count);
445        let mut pos = 2;
446
447        for _ in 0..column_count {
448            if pos + 4 > payload.len() {
449                return Err("DataRow truncated".to_string());
450            }
451
452            let len = i32::from_be_bytes([
453                payload[pos],
454                payload[pos + 1],
455                payload[pos + 2],
456                payload[pos + 3],
457            ]);
458            pos += 4;
459
460            if len == -1 {
461                // NULL value
462                columns.push(None);
463            } else {
464                if len < -1 {
465                    return Err(format!("DataRow invalid column length: {}", len));
466                }
467                let len = len as usize;
468                if len > payload.len().saturating_sub(pos) {
469                    return Err("DataRow column data truncated".to_string());
470                }
471                let data = payload[pos..pos + len].to_vec();
472                pos += len;
473                columns.push(Some(data));
474            }
475        }
476
477        if pos != payload.len() {
478            return Err("DataRow has trailing bytes".to_string());
479        }
480
481        Ok(BackendMessage::DataRow(columns))
482    }
483
484    fn decode_command_complete(payload: &[u8]) -> Result<Self, String> {
485        if payload.last().copied() != Some(0) {
486            return Err("CommandComplete missing null terminator".to_string());
487        }
488        let tag_bytes = &payload[..payload.len() - 1];
489        if tag_bytes.is_empty() {
490            return Err("CommandComplete tag is empty".to_string());
491        }
492        if tag_bytes.contains(&0) {
493            return Err("CommandComplete contains interior null byte".to_string());
494        }
495        let tag = decode_utf8(tag_bytes, "CommandComplete tag")?;
496        Ok(BackendMessage::CommandComplete(tag))
497    }
498
499    fn decode_error_response(payload: &[u8]) -> Result<Self, String> {
500        Ok(BackendMessage::ErrorResponse(Self::parse_error_fields(
501            payload,
502        )?))
503    }
504
505    fn parse_error_fields(payload: &[u8]) -> Result<ErrorFields, String> {
506        if payload.last().copied() != Some(0) {
507            return Err("ErrorResponse missing final terminator".to_string());
508        }
509        let mut fields = ErrorFields::default();
510        let mut i = 0;
511        while i < payload.len() && payload[i] != 0 {
512            let field_type = payload[i];
513            i += 1;
514            let end = payload[i..]
515                .iter()
516                .position(|&b| b == 0)
517                .map(|p| p + i)
518                .ok_or("ErrorResponse field missing null terminator")?;
519            let value = decode_utf8(&payload[i..end], "ErrorResponse field")?;
520            i = end + 1;
521
522            match field_type {
523                b'S' => fields.severity = value,
524                b'C' => fields.code = value,
525                b'M' => fields.message = value,
526                b'D' => fields.detail = Some(value),
527                b'H' => fields.hint = Some(value),
528                _ => {}
529            }
530        }
531        if i + 1 != payload.len() {
532            return Err("ErrorResponse has trailing bytes after terminator".to_string());
533        }
534        Ok(fields)
535    }
536
537    fn decode_parameter_description(payload: &[u8]) -> Result<Self, String> {
538        if payload.len() < 2 {
539            return Err("ParameterDescription payload too short".to_string());
540        }
541        let raw_count = i16::from_be_bytes([payload[0], payload[1]]);
542        if raw_count < 0 {
543            return Err(format!("ParameterDescription invalid count: {}", raw_count));
544        }
545        let count = raw_count as usize;
546        let expected_len = 2 + count * 4;
547        if payload.len() < expected_len {
548            return Err(format!(
549                "ParameterDescription truncated: expected {} bytes, got {}",
550                expected_len,
551                payload.len()
552            ));
553        }
554        let mut oids = Vec::with_capacity(count);
555        let mut pos = 2;
556        for _ in 0..count {
557            oids.push(u32::from_be_bytes([
558                payload[pos],
559                payload[pos + 1],
560                payload[pos + 2],
561                payload[pos + 3],
562            ]));
563            pos += 4;
564        }
565        if pos != payload.len() {
566            return Err("ParameterDescription has trailing bytes".to_string());
567        }
568        Ok(BackendMessage::ParameterDescription(oids))
569    }
570
571    fn decode_copy_in_response(payload: &[u8]) -> Result<Self, String> {
572        if payload.len() < 3 {
573            return Err("CopyInResponse payload too short".to_string());
574        }
575        let format = payload[0];
576        if format > 1 {
577            return Err(format!(
578                "CopyInResponse invalid overall format code: {}",
579                format
580            ));
581        }
582        let num_columns = if payload.len() >= 3 {
583            let raw = i16::from_be_bytes([payload[1], payload[2]]);
584            if raw < 0 {
585                return Err(format!(
586                    "CopyInResponse invalid negative column count: {}",
587                    raw
588                ));
589            }
590            raw as usize
591        } else {
592            0
593        };
594        let mut column_formats = Vec::with_capacity(num_columns);
595        let mut pos = 3usize;
596        for _ in 0..num_columns {
597            if pos + 2 > payload.len() {
598                return Err("CopyInResponse truncated column format list".to_string());
599            }
600            let raw = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
601            if !(0..=1).contains(&raw) {
602                return Err(format!("CopyInResponse invalid format code: {}", raw));
603            }
604            column_formats.push(raw as u8);
605            pos += 2;
606        }
607        if pos != payload.len() {
608            return Err("CopyInResponse has trailing bytes".to_string());
609        }
610        Ok(BackendMessage::CopyInResponse {
611            format,
612            column_formats,
613        })
614    }
615
616    fn decode_copy_out_response(payload: &[u8]) -> Result<Self, String> {
617        if payload.len() < 3 {
618            return Err("CopyOutResponse payload too short".to_string());
619        }
620        let format = payload[0];
621        if format > 1 {
622            return Err(format!(
623                "CopyOutResponse invalid overall format code: {}",
624                format
625            ));
626        }
627        let num_columns = if payload.len() >= 3 {
628            let raw = i16::from_be_bytes([payload[1], payload[2]]);
629            if raw < 0 {
630                return Err(format!(
631                    "CopyOutResponse invalid negative column count: {}",
632                    raw
633                ));
634            }
635            raw as usize
636        } else {
637            0
638        };
639        let mut column_formats = Vec::with_capacity(num_columns);
640        let mut pos = 3usize;
641        for _ in 0..num_columns {
642            if pos + 2 > payload.len() {
643                return Err("CopyOutResponse truncated column format list".to_string());
644            }
645            let raw = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
646            if !(0..=1).contains(&raw) {
647                return Err(format!("CopyOutResponse invalid format code: {}", raw));
648            }
649            column_formats.push(raw as u8);
650            pos += 2;
651        }
652        if pos != payload.len() {
653            return Err("CopyOutResponse has trailing bytes".to_string());
654        }
655        Ok(BackendMessage::CopyOutResponse {
656            format,
657            column_formats,
658        })
659    }
660
661    fn decode_copy_both_response(payload: &[u8]) -> Result<Self, String> {
662        if payload.len() < 3 {
663            return Err("CopyBothResponse payload too short".to_string());
664        }
665        let format = payload[0];
666        if format > 1 {
667            return Err(format!(
668                "CopyBothResponse invalid overall format code: {}",
669                format
670            ));
671        }
672        let num_columns = if payload.len() >= 3 {
673            let raw = i16::from_be_bytes([payload[1], payload[2]]);
674            if raw < 0 {
675                return Err(format!(
676                    "CopyBothResponse invalid negative column count: {}",
677                    raw
678                ));
679            }
680            raw as usize
681        } else {
682            0
683        };
684        let mut column_formats = Vec::with_capacity(num_columns);
685        let mut pos = 3usize;
686        for _ in 0..num_columns {
687            if pos + 2 > payload.len() {
688                return Err("CopyBothResponse truncated column format list".to_string());
689            }
690            let raw = i16::from_be_bytes([payload[pos], payload[pos + 1]]);
691            if !(0..=1).contains(&raw) {
692                return Err(format!("CopyBothResponse invalid format code: {}", raw));
693            }
694            column_formats.push(raw as u8);
695            pos += 2;
696        }
697        if pos != payload.len() {
698            return Err("CopyBothResponse has trailing bytes".to_string());
699        }
700        Ok(BackendMessage::CopyBothResponse {
701            format,
702            column_formats,
703        })
704    }
705
706    fn decode_notification_response(payload: &[u8]) -> Result<Self, String> {
707        if payload.len() < 6 {
708            // Minimum: 4 (process_id) + 1 (channel NUL) + 1 (payload NUL)
709            return Err("NotificationResponse too short".to_string());
710        }
711        let process_id = i32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
712
713        // Channel name (null-terminated)
714        let mut i = 4;
715        let remaining = payload.get(i..).unwrap_or(&[]);
716        let channel_end = remaining
717            .iter()
718            .position(|&b| b == 0)
719            .ok_or("NotificationResponse: missing channel null terminator")?;
720        if channel_end == 0 {
721            return Err("NotificationResponse channel is empty".to_string());
722        }
723        let channel = decode_utf8(&remaining[..channel_end], "NotificationResponse channel")?;
724        i += channel_end + 1;
725
726        // Payload (null-terminated)
727        let remaining = payload.get(i..).unwrap_or(&[]);
728        let payload_end = remaining
729            .iter()
730            .position(|&b| b == 0)
731            .ok_or("NotificationResponse: missing payload null terminator")?;
732        let notification_payload =
733            decode_utf8(&remaining[..payload_end], "NotificationResponse payload")?;
734        if i + payload_end + 1 != payload.len() {
735            return Err("NotificationResponse has trailing bytes".to_string());
736        }
737
738        Ok(BackendMessage::NotificationResponse {
739            process_id,
740            channel,
741            payload: notification_payload,
742        })
743    }
744}
qail_pg/protocol/wire/backend.rs

qail_pg/protocol/wire/
backend.rs
