Раздел 10 · Databases

Своя СУБД на Rust · урок 11: бинарный протокол и сервер

senior~110 мин

без входаВойди, чтобы отмечать прогресс, вести заметки и решать задачи в редакторе. войти

Своя СУБД на Rust · урок 11: бинарный протокол и сервер

Кому это. За десять уроков мы собрали целый движок на Rust: страницы, буферный пул с PageGuard, кучу, B+Tree, парсер SQL, планировщик и исполнитель, журнал с CRC-32, транзакции и MVCC, блокировки с поиском дедлока, vacuum. Но говорить с движком умеют только тесты в том же процессе.

Зачем. Это финал воркшопа: выставляем движок в сеть. Придумаем своё кадрирование, реализуем simple query protocol в духе Postgres, напишем сессию, сервер и клиента. В конце клиент откроет сокет, пошлёт SELECT, получит типизированные строки, а BEGIN и COMMIT заработают по сети. Байтовый контракт общий для всех языков воркшопа, поэтому наш клиент поговорит с сервером на TS или Go, и наоборот.

Зачем именно Rust. Здесь собираются вместе все решения, которые мы откладывали. Сервер синхронный, на std::net, без асинхронного рантайма: курс собирает движок без крейтов. Поток на accept и поток на соединение, а Database лежит в Arc<Mutex<Database>>: движок однопоточный, и мьютекс выполняет запросы по одному. Тут окупается FlushHooks: Send из урока 7: без него Mutex<Database> нельзя было бы отдать в thread::spawn. Ошибки движка это Result, и на границе протокола они превращаются в ErrorResponse с SQLSTATE-кодом через Error::sqlstate(). Тесты слушают 127.0.0.1:0, и ОС сама выдаёт свободный порт.

Важно про объём. Урок аддитивный: новый модуль src/wire/ из пяти файлов, движок не меняется. Кадрирование и кодеки сообщений чистые, без ввода-вывода. Сессия превращает SQL в сообщения и тестируется без сети. Сервер и клиент это тонкий слой поверх сокетов.

Где мы в движке

┌────────────────────────────────────────────────┐
│  Бинарный протокол + TCP-сервер                  │  урок 11  ← мы здесь
├────────────────────────────────────────────────┤
│  Исполнитель (Volcano) + EXPLAIN                 │  урок 6
│  Планировщик, binder, каталог                    │  урок 6
│  Парсер SQL (рекурсивный спуск + Pratt)          │  урок 5
├────────────────────────────────────────────────┤
│  Vacuum                                          │  урок 10
│  MVCC (видимость, версии кортежа)                │  урок 8
│  Блокировки (2PL), конкуренция                   │  урок 9
│  Менеджер транзакций (xid, состояния)            │  урок 8
│  WAL (журнал, redo, recovery)                    │  урок 7
├────────────────────────────────────────────────┤
│  B+Tree индекс                                   │  урок 4
│  Куча (slotted pages, seq scan)                  │  урок 3
├────────────────────────────────────────────────┤
│  Буферный пул (кэш страниц, вытеснение)           │  урок 2
│  DiskManager (страницы поверх файла)              │  урок 1
└────────────────────────────────────────────────┘

Идея протокола

Клиент и сервер обмениваются сообщениями. Каждое сообщение это один байт типа и payload. Клиент шлёт Query (SQL-строка) или Terminate (до свидания). Сервер на каждый Query отвечает серией: для SELECT это RowDescription, сколько угодно DataRow, CommandComplete с тегом вроде SELECT 2 и ReadyForQuery со статусом транзакции. Ошибка это ErrorResponse с кодом и текстом, а за ним всё равно ReadyForQuery. Клиент читает ответ до ReadyForQuery и только потом шлёт следующий запрос.

Типы сообщений это ASCII-буквы, как в Postgres: Q, X, T, D, C, E, Z. Числа на проводе big-endian, в сетевом порядке, в отличие от little-endian на диске. Два порядка байт в одном движке это обычное дело: формат файла выбирают под процессор, формат провода под традицию сетевых протоколов.

Кадрирование

// src/wire/framing.rs
//! Кадр: u8 тип, u32 длина big-endian (включает 4 байта длины, не включает байт типа),
//! payload длиной length - 4.

use crate::error::{invalid, Result};

/// Размер заголовка кадра: байт типа плюс u32 длины.
const FRAME_HEADER_SIZE: usize = 5;

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Frame {
    pub ty: u8,
    pub payload: Vec<u8>,
}

/// Собрать кадр: байт типа, длина (4 плюс payload), payload.
pub fn encode_frame(ty: u8, payload: &[u8]) -> Vec<u8> {
    let mut frame = Vec::with_capacity(FRAME_HEADER_SIZE + payload.len());
    frame.push(ty);
    frame.extend_from_slice(&(4 + payload.len() as u32).to_be_bytes());
    frame.extend_from_slice(payload);
    frame
}

#[derive(Debug, Default)]
pub struct FrameReader {
    buffer: Vec<u8>,
}

impl FrameReader {
    pub fn new() -> FrameReader {
        FrameReader::default()
    }

    /// Добавить чанк и вернуть все кадры, собравшиеся целиком. Длина меньше 4 это
    /// мусор на входе: такой кадр не может существовать.
    pub fn push(&mut self, chunk: &[u8]) -> Result<Vec<Frame>> {
        self.buffer.extend_from_slice(chunk);
        let mut frames = Vec::new();
        let mut start = 0;
        while self.buffer.len() - start >= FRAME_HEADER_SIZE {
            let length_bytes = self.buffer[start + 1..start + FRAME_HEADER_SIZE]
                .try_into()
                .expect("4 bytes");
            let length = u32::from_be_bytes(length_bytes) as usize;
            if length < 4 {
                return Err(invalid(format!("frame length {length} is below 4")));
            }
            let total = 1 + length;
            if self.buffer.len() - start < total {
                break; // кадр ещё не дочитан
            }
            frames.push(Frame {
                ty: self.buffer[start],
                payload: self.buffer[start + FRAME_HEADER_SIZE..start + total].to_vec(),
            });
            start += total;
        }
        self.buffer.drain(..start);
        Ok(frames)
    }
}

Длина в кадре включает свои 4 байта, но не байт типа: так устроен протокол Postgres, и наш кадр байт в байт совпадает с его форматом. Поэтому длина меньше 4 невозможна, и FrameReader отвергает её как мусор.

FrameReader это маленький конечный автомат над буфером. push дописывает чанк, режет из начала буфера все целые кадры и оставляет хвост до следующего чанка. Один read из сокета может принести полкадра, три кадра или два с половиной, и push справится с любым вариантом. Разберём Rust-детали.

  1. self.buffer[start + 1..start + 5].try_into() превращает срез &[u8] в массив [u8; 4] для u32::from_be_bytes. Преобразование может не удаться, если длина среза не 4. Здесь она 4 по построению, поэтому .expect("4 bytes") фиксирует инвариант, а не обрабатывает ошибку.
  2. Курсор start сдвигается по кадрам, а буфер чистится один раз в конце через drain(..start). Удаление из начала Vec после каждого кадра сдвигало бы хвост снова и снова; один drain двигает его единожды.
  3. Длина приходит из сети и проверяется до использования. length < 4 это ошибка, а огромная длина просто оставит кадр недочитанным. Ограничение сверху на размер кадра стоит добавить в настоящем сервере, иначе клиент заставит нас копить буфер бесконечно; это часть упражнения про handshake.

Сообщения

// src/wire/messages.rs
//! Simple query protocol в духе Postgres, числа big-endian. DataRow на проводе это
//! текст: int десятичной строкой, bool как t/f, text как есть.

/// Байты типа сообщений (ASCII).
pub mod msg {
    pub const QUERY: u8 = b'Q';
    pub const TERMINATE: u8 = b'X';
    pub const ROW_DESCRIPTION: u8 = b'T';
    pub const DATA_ROW: u8 = b'D';
    pub const COMMAND_COMPLETE: u8 = b'C';
    pub const ERROR_RESPONSE: u8 = b'E';
    pub const READY_FOR_QUERY: u8 = b'Z';
}

/// Статус транзакции в ReadyForQuery.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TxnStatus {
    Idle,
    InTransaction,
    Failed,
}

impl TxnStatus {
    pub fn byte(self) -> u8 {
        match self {
            TxnStatus::Idle => b'I',
            TxnStatus::InTransaction => b'T',
            TxnStatus::Failed => b'E',
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ClientMessage {
    Query(String),
    Terminate,
}

impl ClientMessage {
    /// Query это SQL плюс завершающий NUL, Terminate пуст.
    pub fn encode_payload(&self) -> Vec<u8> {
        match self {
            ClientMessage::Query(sql) => {
                let mut payload = Vec::with_capacity(sql.len() + 1);
                payload.extend_from_slice(sql.as_bytes());
                payload.push(0);
                payload
            }
            ClientMessage::Terminate => Vec::new(),
        }
    }

    pub fn decode(ty: u8, payload: &[u8]) -> Result<ClientMessage> {
        match ty {
            msg::QUERY => {
                let sql = payload.strip_suffix(&[0]).unwrap_or(payload);
                Ok(ClientMessage::Query(
                    String::from_utf8_lossy(sql).into_owned(),
                ))
            }
            msg::TERMINATE => Ok(ClientMessage::Terminate),
            other => Err(invalid(format!(
                "неизвестный тип клиентского сообщения 0x{other:x}"
            ))),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ServerMessage {
    RowDescription(Vec<Field>),
    /// Ячейки текстом, None это NULL (длина -1).
    DataRow(Vec<Option<String>>),
    CommandComplete(String),
    ErrorResponse {
        code: String,
        message: String,
    },
    ReadyForQuery(TxnStatus),
}

Сообщения это два enum, клиентский и серверный. Каждый вариант несёт ровно свои данные, и match по сообщению не даст забыть вид ответа. b'Q' это байтовый литерал: u8 со значением ASCII-кода буквы. Модуль msg повторяет приём с кодами записей журнала: на проводе голый байт, и незнакомый тип должен стать ошибкой разбора, а не паникой.

strip_suffix(&[0]) снимает завершающий NUL, если он есть, а unwrap_or(payload) берёт payload как есть, если его нет. String::from_utf8_lossy заменяет битые последовательности символом замены вместо ошибки: SQL с битым UTF-8 всё равно не разберётся парсером, и он вернёт понятную синтаксическую ошибку. 0x{other:x} это шестнадцатеричный вывод прямо в строке формата.

Кодирование серверных сообщений собирает payload по варианту:

// src/wire/messages.rs (продолжение, в impl ServerMessage)

    pub fn encode_payload(&self) -> Vec<u8> {
        let mut out = Vec::new();
        match self {
            ServerMessage::RowDescription(fields) => {
                out.extend_from_slice(&(fields.len() as u16).to_be_bytes());
                for field in fields {
                    out.extend_from_slice(&(field.name.len() as u16).to_be_bytes());
                    out.extend_from_slice(field.name.as_bytes());
                    out.push(type_code(field.ty));
                }
            }
            ServerMessage::DataRow(cells) => {
                out.extend_from_slice(&(cells.len() as u16).to_be_bytes());
                for cell in cells {
                    match cell {
                        None => out.extend_from_slice(&(-1i32).to_be_bytes()),
                        Some(text) => {
                            out.extend_from_slice(&(text.len() as i32).to_be_bytes());
                            out.extend_from_slice(text.as_bytes());
                        }
                    }
                }
            }
            ServerMessage::CommandComplete(tag) => out.extend_from_slice(tag.as_bytes()),
            ServerMessage::ErrorResponse { code, message } => {
                out.extend_from_slice(&(code.len() as u16).to_be_bytes());
                out.extend_from_slice(code.as_bytes());
                out.extend_from_slice(message.as_bytes());
            }
            ServerMessage::ReadyForQuery(status) => out.push(status.byte()),
        }
        out
    }

DataRow несёт ячейку как Option<String>, а на проводе длину i32: -1 значит NULL, как в Postgres. Знаковый тип длины здесь ради этого сентинела. При разборе обратная операция тоже красивая: usize::try_from(len) для отрицательной длины даёт Err, и ветка Err(_) => None превращает её в NULL без отдельного сравнения с -1.

Значения на проводе текстом: int десятичной строкой, bool как t и f, text как есть. Клиент разбирает их обратно по типу из RowDescription:

// src/wire/messages.rs (продолжение)

/// Текстовое представление значения для DataRow (None это NULL).
pub fn encode_value_text(value: &Value) -> Option<String> {
    match value {
        Value::Null => None,
        Value::Int(int) => Some(int.to_string()),
        Value::Bool(flag) => Some(if *flag { "t" } else { "f" }.to_owned()),
        Value::Text(text) => Some(text.clone()),
    }
}

/// Разобрать текстовое значение колонки по её типу.
pub fn decode_value_text(ty: ColumnType, text: &str) -> Result<Value> {
    Ok(match ty {
        ColumnType::Int => Value::Int(
            text.parse()
                .map_err(|_| invalid(format!("не число в int-колонке: {text}")))?,
        ),
        ColumnType::Bool => Value::Bool(text == "t"),
        ColumnType::Text => Value::Text(text.to_owned()),
    })
}

text.parse() выводит целевой тип из Value::Int(i64) и разбирает строку в i64. Число за пределами i64 на проводе станет ошибкой разбора, а не тихим переполнением. Здесь видна одна из сознательных развилок порта с TS-эталоном. В TS значение INT живёт в bigint без верхней границы, и v + 1 у максимального числа даёт честно большее число, а выход за 64 бита всплывает только при сериализации кортежа. В Rust Value::Int это i64, и вычислитель урока 6 считает через checked_add и его родню: переполнение становится Error::Engine("integer overflow") уже в выражении. По сети клиент получит его как ErrorResponse с кодом 42000.

decode серверного сообщения читает payload через маленький курсор Reader с методами take, u16 и text. Каждый проверяет границы через bytes::slice и возвращает Result: payload пришёл из сети, и короткое сообщение от неисправного сервера должно стать ошибкой, а не паникой клиента.

Ошибки становятся кодами

Сессия на каждую ошибку движка отвечает ErrorResponse. Код выбирает Error::sqlstate() из урока 9:

Вариант ErrorSQLSTATEИмя в Postgres
Uniqueness23505unique_violation
Serialization40001serialization_failure
Deadlock40P01deadlock_detected
LockWait55P03lock_not_available
Sql42601syntax_error
Engine42000syntax_error_or_access_rule_violation
Io, InvalidXX000internal_error

Сессия добавляет ещё один код, 25P02 (in_failed_sql_transaction): команда внутри упавшей транзакции. Клиент сам порождает 08006 (connection_failure), когда сервер закрыл сокет.

Особенно полезен 55P03 по сети. В уроке 9 мы решили, что менеджер блокировок не ждёт. Теперь у этого решения есть клиентская сторона: второе соединение, которое тронуло строку под чужим локом, получает 55P03 сразу, а не висит. Клиент вправе повторить оператор позже.

дальше после входа

Урок целиком открыт всем, кто вошёл. Войди по своему email, чтобы читать дальше.

Вход пока по приглашению. Если хочешь учиться, напиши Vassiliy.

войти