Канал реального времени

Эта страница переведена с английского. Читать страницу на английском

Запросы Zeyn не хранят состояния: сервер не держит ни сокета, ни состояния экрана на пользователя. Некоторым экранам нужно больше: двое редактируют одну диаграмму или список должен обновляться, когда кто-то другой сохраняет данные. Cargo feature realtime добавляет для этого опциональный канал WebSocket. Приложения, не включающие её, ничего из этого не компилируют, а страницы, которые её не используют, ничего не загружают.

WebSocket часто блокируется в корпоративных сетях, поэтому относитесь к каналу как к улучшению: каждый экран должен по-прежнему работать без него (редактор диаграмм всё равно сохраняет через своё действие; список всё равно обновляется, когда пользователь открывает его снова).

1. Включение

zeyn = { version = "...", features = ["realtime"] }

let realtime = Realtime::new()
    .documents("diagram", "diagram.read", "diagram.write", MyDocStore::new(pool))
    .topic("orders", "sales.order.read")
    .limits(Limits::default());
let hub = realtime.hub();          // clone it into stores that notify
App::new(..).realtime(realtime)    // also registers `Hub` as a service

2. Конечные точки

Каждое рукопожатие проверяется до обновления протокола, в таком порядке:

1. Канал включён (иначе 404).
2. Origin: заголовок Origin должен называть хост, на который пришёл запрос, или origin, добавленный через Realtime::allow_origin. Браузеры всегда отправляют его при рукопожатиях WebSocket; отсутствующий отклоняется. Именно это не даёт другому сайту открыть сокет с cookie пользователя.
3. CSRF-токен: браузеры не могут задавать заголовки для WebSocket, поэтому токен из meta[name=zeyn-csrf] передаётся в параметре запроса csrf и сравнивается за постоянное время с cookie zeyn_csrf, точно так же, как x-zeyn-csrf для POST. Сбои дают 403.
4. Идентичность: тот же аутентификатор, что и для каждого запроса; пользователь должен быть вошедшим в систему (403).
5. Лимит на пользователя (429).
6. Документы: объявленный kind и простой id (от 1 до 64 букв, цифр, -, _, .), иначе 404; разрешение на чтение для этого kind (403). Темы: каждая тема объявлена, а её permission есть у пользователя (403).
7. Interceptors: interceptors приложения выполняются с вызовом вида Live (screen это kind документа, для тем пусто), поэтому общее для приложения правило, такое как RequireMfa или проверка заблокированной учётной записи, охватывает и живые каналы. Отказ отвечает своим статусом.
8. Лимиты комнат: комнаты в памяти и клиенты в комнате (503 с Retry-After).

Разрешение на запись проверяется один раз на соединение: клиент без него получает обновления, но закрывается с кодом 1008 в тот момент, когда отправляет своё.

3. Документы и Y-CRDT

Документы это документы Y-CRDT, которые хранятся с помощью yrs. Комнаты идентифицируются по арендатору, kind и id документа, поэтому два арендатора никогда не делят комнату, даже при одинаковых id.

Кадры бинарные. Первый байт это тег; 1 (FRAME_UPDATE) несёт обновление Y в кодировке v1.

• При подключении сервер отправляет весь документ одним обновлением.
• Клиент отправляет обновления для своих изменений; сервер применяет каждое (кадр, не являющийся корректным обновлением, закрывает сокет с 1008) и передаёт те же байты каждому другому клиенту комнаты.
• После каждого (пере)подключения остров отправляет обратно всё своё состояние, поэтому правки, сделанные офлайн, сливаются. Обновления Y идемпотентны и коммутативны, поэтому порядок и повторы не важны.

Сохранение идёт через трейт:

pub trait DocStore: Send + Sync + 'static {
    fn load(&self, key: RoomKey) -> BoxFuture<Result<Option<Vec<u8>>>>;
    fn save(&self, key: RoomKey, state: Vec<u8>) -> BoxFuture<Result<()>>;
}

load выполняется при открытии комнаты (ошибка отклоняет соединение с 503, и так же хранилище ограничивает, какие документы существуют). save получает весь документ одним обновлением и выполняется каждые persist_every, пока у документа есть изменения, и перед тем, как простаивающая комната будет выгружена. Неудавшееся сохранение повторяется в следующем раунде, а комната сохраняется. MemoryDocStore предназначен для тестов и демо.

Со стороны острова zeyn-island с feature realtime предоставляет live::Channel:

let channel = Channel::document("diagram", "main", on_open, on_update, on_state)?;
channel.send_update(&update);   // after a local change
channel.close();                // in unmount

Он переподключается через 1 секунду, удваивая интервал до 30 секунд, и останавливается после close или 1008.

4. Темы: экраны, обновляющиеся на месте

Live::new("orders").add(DataGrid::of(search_orders)...)

#[action(permission = "sales.order.write")]
pub async fn save_order(o: Valid<Order>, store: Orders, hub: Hub, ctx: Ctx) -> Result<Order> {
    let saved = store.save(o.into_inner())?;
    hub.notify(&ctx, "orders");
    Ok(saved)
}

Live отображает div[data-zeyn-live="orders"]. Страница, чей экран содержит такой элемент, подключает /_zeyn/live.js (экран, приходящий через nav.js, получает x-zeyn-live: 1, и скрипт импортируется тогда). Скрипт открывает один сокет для всех тем на странице. При уведомлении каждый подписанный элемент перезагружает свои сетки на месте (zeyn:reload, текущая страница сетки, сортировка и фильтры) или весь экран на месте (zeyn:refresh), когда сеток в нём нет. Уведомления в пределах 300 мс обрабатываются один раз.

Уведомления несут только имя темы, никогда не данные: обновление идёт через обычную конечную точку сетки со всеми проверками разрешений. notify достигает только арендатора вызывающего; notify_tenant(Some(id), topic) называет конкретного, None означает приложение без мультиарендности.

5. Память и обратное давление

Каждый фрагмент состояния ограничен (Limits, значения по умолчанию в скобках):

Рассылки никогда не ждут: обновление предлагается ограниченной очереди каждого клиента через try_send. Клиент, у которого очередь заполнена, сразу удаляется из комнаты, поэтому один медленный читатель не может удерживать память или замедлять остальных; он переподключается и получает весь документ. Подписчик темы с заполненной очередью лишь пропускает это уведомление (следующее всё равно его обновит).

Поэтому худший случай по памяти это примерно rooms * doc_bytes для документов плюс clients * queue * message_bytes для очередей (обновление разделяется между очередями, а не копируется). Сервер showcase с двумя живыми вкладками на диаграмме использует в сумме около 19 МБ (отладочная сборка). Hub::stats() сообщает комнаты, клиентов, подписчиков и байты документов для мониторинга.

6. Замечания по безопасности

• Канал никогда не выполняет действия: документы это данные, которые приложение читает через свой DocStore, и именно там оно их проверяет (showcase ограничивает каждую сохраняемую диаграмму через Diagram::clamped).
• Документ доступен для чтения любому, у кого есть разрешение на чтение этого kind в арендаторе; для документов с разной аудиторией используйте отдельные kinds (или id, которые хранилище отклоняет).
• Политика каждой страницы разрешает WebSocket к собственному origin приложения через default-src 'self'; ничто в content security policy ради канала не смягчается.

7. Пример

examples/showcase: откройте Islands, Diagram editor в двух вкладках. Перемещение или добавление фигуры в одной вкладке появляется в другой, а сетка «Shapes stored on the server» ниже обновляется на месте примерно за две секунды, когда документ сохраняется через DocStore и уведомляется тема diagram.

Ограничения текущей версии: комнаты живут в одном процессе (нескольким экземплярам сервера нужны sticky sessions или общий брокер), присутствие (кто ещё здесь, курсоры) не встроено, а документ отправляется целиком при каждом подключении.