Архитектура SDK для работы с YDB-топиками
Эта статья - черновик заметок по разработке SDK, может произвольным образом меняться, дополняться и удаляться. Версия НЕ стабильна.
Отправка сообщений сообщений
Описание протокола отправки на уровне GRPC-сообщений
Запись сообщения в транзакции на уровне SDK
С точки зрения API для пользователя для каждой транзакции создаётся отдельный транзакционный писатель. Все сообщения, отправленные через этого писателя будут опубликованы с коммитом транзакции. При завершении транзакции по любой причине - писатель прекращает свою работу и на все попытки записи возвращает ошибку.
Наличие отдельного транзакционного писателя решает проблему обработки ошибок. Например может получиться ситуация, когда клиент считает что транзакция есть, а на самом деле транзакции уже нет (нода ребутнулась). Тогда при попытке отправить сообщение в несуществующую транзакцию клиент получит неретраибельную ошибку и прекратит свою работу. Внутри транзакции это нормально - т.к. транзакции нет и клиент работу должен прекратить. При этом другие транзакции могут продолжать работать и это не должно на них влиять. Идеологически писатель гарантирует сохранность и порядок всех сообщений и не допускает пропусков при отправке. Если бы мы оставили работу с единым писателем, то из этой логики пришлось бы делать исключения. Наличие отдельного писателя сильно упрощает эту часть: для клиента это выглядит как отдельный писатель к транзакции. Когда он получил ошибку - писатель больше не работает. При этом для эффективной работы внутри SDK может быть пул писателей, который будет переиспользоваться параллельными транзакциями.
По умолчанию писатель создаётся без привязки к конкретной партиции, без ProducerID, аналогично ошибкам при работе с таблицами.
ВАЖНОЕ ПРО РЕТРАИ: если писатель создан без ProducerID (рекомендуемый вариант, поля InitRequest.producer_id и MessageData.seq_no - не заполняются), то операция записи становится НЕ идемпотентной и это нужно учитывать при принятии решений о ретрае внутри самого писателя.
Например grpc-unavailable ретраить уже нельзя, т.к. producer_id нет, seq_no - нет и дедупликация на сервере работать не будет. И если поретраить отправку сообщения, которое могло уже сохраниться на сервере - получится дубль, а этого допускать нельзя. Ошибки, безопасные для ретрая - например серверный unavailable или overloaded ретраить можно, но это оптимизация. В простом случае можно фейлиться на любой ошибке и тогда транзакция поретраится общим ретраером (будет больше работы на клиенте, но правильность результата сохраняется и если ошибки редкие, то это не страшно).
На сервер нельзя писать параллельно с одним и тем же producer_id, транзакции обычно идут параллельно, поэтому единый producer_id на процесс уже не подойдёт. Использование случайного producer_id для каждой транзакции будет засорять пространство их имён и заставлять сервер делать лишнюю работу по их проверке.
Внутри SDK может переиспользоваться пул писателей. При этом важно клиенту отдавать надстройки над писателями, а не сам объект писателя - чтобы предотвратить возможность его переиспользования между транзакциями (и защитить пользователя от незаметных ошибок). Тут стоит особенно аккуратно подойти к реализации обработки ошибок, т.к. писатели асинхронные и ошибку можно получить уже после отправки сообщения - когда писатель уже перевыдан другой транзакции. Нужно убедиться что по предыдущим сообщениям ошибок уже не будет (например переиспользщовать писателя только если получены все подтверждения и транзакция закоммитилась).
При вызове Writer.Write управление пользователю возвращается сразу, а сообщение складывается в буфер и может быть отправлено позже - вместе с другими сообщениями, которые могут придти позже. Тут можно переиспользовать механизм буферизации сообщений из обычного писателя. Для гарантии того что все сообщения будут доставлены на сервер перед коммитом - внутри SDK происходит подписка писателя на события транзакции. Транзакция отслеживает момент своего завершения (по ошибке) или шага с коммитом (как явный коммит, так и отправка флажка коммита вместе с запросом). А писатель на эти события реагирует: перед коммитом дожидается подтверждения о записи всех сообщений от сервера. Если произошла ошибка - фейлит транзакцию. При завершении транзакции - писатель закрывается или отправляется в пул для переиспользования.
Api транзакционного писателя должно повторять API обычного писателя на отправку сообщений. При этом желательно исключить способы дождаться от сервера подтверждения о записи сообщения, т.к. это всегда происходит внутри SDK перед коммитом и SDK сам гарантирует что все сообщения, положенные в его буфер до коммита будут переданы на сервер и закоммитятся вместе с транзакцией.
GRPC-протокол
С точки зрения протокола отправка сообщения в транзакции - это Простая отправка сообщений + заполнение поля WriteRequest.tx, в качестве подтверждения будет приходить WriteAck.WrittenInTx, в котором не будет Offset-а, т.к. он будет известен только при коммите транзакции.
Отправка батча с сообщениями на сервер
При отправке пачки сообщений на сервер важно учитывать лимит на размер одного grpc-сообщения. В случае, если скорость записи сообщений в буфер превышает скорость отправки, а размер буфера достаточно большой - может получиться ситуация, когда при сериализации буфера размер grpc-сообщения превысит лимит на размер grpc-сообщения, установленные на клиентской или серверной стороне в grpc-библиотеке. Эту ситуацию нужно проверять и когда происходит превышение - разбивать большое сообщение на несколько поменьше - чтобы они проходили в ограничения grpc.
Закрытие писателя
При вызове метода закрытия (Close) писатель переходит в режим закрытия и больше не принимает новых сообщений. На все попытки вызвать любые методы нужно возвращать ошибку, о том что работа писателя завершена. Можно разрешить повторное закрытие писателя без ошибок - это зависит от организации остального SDK и обычаев языка программирования.
При остановке писателя нужно дождаться подтверждения от сервера о записи всех сообщений. При этом внутри писатель может переподключаться к серверу, ретраить ошибки и т.п. Ожидание останавливается когда получены подтверждения для всех сообщений, которые писатель успел принять до вызова метода остановки или при получении НЕРЕТРАИБЕЛЬНОЙ ошибки. Опционально может задаваться таймаут ожидания, тогда ожидание ограничивается этим таймаутом - по его истечении все фоновые процессы принудительно останавливаются.
Возврат из метода остановки делается после завершения всех фоновых процессов.
Чтение сообщений из топика
Остановка читателя
При вызове метода остановки читатель переходит в режим закрытия и больше не обрабатывает запросы на чтение и коммиты сообщений. На все вызовы любых методов нужно возвращать ошибку о том, что работа читателя уже остановлена. Можно разрешить повторное закрытие читателя без ошибок - это зависит от организации остального SDK и обычаев языка программирования.
При остановке читателя нужно дождать подтверждения о коммитах, принятых до вызова метода остановки. Ожидание останавливается при получении подтверждения обо всех коммитах, принятых читателем до вызова метода закрытия или после завершения стрима по любой причине, переподключаться НЕ нужно. Коммиты сообщений возможны только внутри стрима, где они были прочитаны, поэтому переподключаться смысла нет - если коммиты не успели дойти до сервера - сообщения нужно будет получить и обработать заново.