Жизненный цикл

SqlxBroker::from_url записывает конфигурацию, connect строит пул и определяет диалект, каждая подписка при старте подготавливает свой SQL, а завершение работы останавливает захваты.

Схема текстом

1. Брокер: SqlxBroker ::from_url(..); ConnectedSqlx Broker; ClosedSqlxBroker locks_released, connections_closed
2. В базе данных: Параметры пула, маршруты: без I/O; Пул построен; взято соединение, диалект определён; ещё одно при listen_notify; Каждая подписка: проверка объявления, построение операторов, чтение версии сервера, подготовка каждого; Идут циклы захвата; аренды продлеваются каждые полсрока; Захваты и продление стоят; блокировки ключей сняты; слушатель и построенный пул закрыты
3. SqlxBroker ::from_url(..) → ConnectedSqlx Broker: connect(self)
4. ConnectedSqlx Broker → ClosedSqlxBroker locks_released, connections_closed: shutdown(self)
5. Параметры пула, маршруты: без I/O → Пул построен; взято соединение, диалект определён; ещё одно при listen_notify
6. Пул построен; взято соединение, диалект определён; ещё одно при listen_notify → Каждая подписка: проверка объявления, построение операторов, чтение версии сервера, подготовка каждого
7. Каждая подписка: проверка объявления, построение операторов, чтение версии сервера, подготовка каждого → Идут циклы захвата; аренды продлеваются каждые полсрока
8. Идут циклы захвата; аренды продлеваются каждые полсрока → Захваты и продление стоят; блокировки ключей сняты; слушатель и построенный пул закрыты
9. Издатель, переживший брокер, отвечает SqlxBrokerError::Closed, а не молчаливым успехом

from_url, from_connect_options, route, poll_interval, lease, connection_limit и listen_notify не выполняют ввода-вывода, поэтому билдер сервиса остаётся синхронным и работает до запуска рантайма. SqlxBroker::new принимает пул, который сервис построил сам, внутри рантайма, и делит с собственным кодом.

Подключение строит пул по записанным параметрам, затем берёт одно его соединение и так проверяет, что база данных доступна: URL, который драйвер не может прочитать, прерывает старт с ошибкой SqlxBrokerError::Url, а недоступная база данных - с ошибкой SqlxBrokerError::Connect. На этом соединении брокер определяет свой диалект, что важно для AnyPool: именно там он узнаёт, к какой базе данных обращается. Затем он возвращает соединение. При listen_notify он берёт ещё одно и держит его до завершения работы, чтобы слушать уведомления.

Подписки открываются по порядку. Сначала проверяется объявление подписки: max_attempts без dead_letter или наоборот, а также предел попыток на таблице без столбца attempt останавливают её с ошибкой SqlxBrokerError::Declaration. Её операторы строятся по описанию таблицы, а вторая подписка на ту же очередь на этом брокере отклоняется. Затем на одном соединении диалект читает версию сервера там, где у него есть нижняя граница версии, как у MySQL и MariaDB, и каждый оператор подготавливается на сервере: таблица или столбец, которых нет в базе данных, останавливают сервис при старте с ошибкой SqlxBrokerError::Schema, которая называет таблицу и оператор. Наконец, при listen_notify подписка начинает слушать канал своей таблицы. Любая ошибка на этом этапе - ошибка старта, до захвата первого сообщения.

Пока сервис работает, каждая подписка захватывает так, как описывает страница Пробуждение, а в форме аренды отдельная задача продлевает каждую аренду в работе раз в половину срока аренды.

Завершение работы поднимает флаг закрытия брокера, и этот флаг отклоняет каждого выданного раньше издателя, ведь общий с сервисом пул остаётся открытым. Оно останавливает циклы захвата и задачи продления аренды. Доставка в работе в форме блокировки строк или аренды завершается как прежде: её транзакция или аренда по-прежнему держит строку, а аренда больше не продлевается. В форме рекомендательной блокировки завершение работы снимает блокировку каждой доставки в работе, а доставка со снятой блокировкой больше не завершается: её завершение возвращает ошибку Closed, и строка возвращается в очередь. Захват, который среда выполнения прервала при остановке подписки, может ещё выполняться на сервере. Его соединение дожидается, пока сервер закончит оператор захвата и откатит его транзакцию, и закрывается. Завершение работы ждёт эти соединения и каждый сеанс, который ещё закрывается, закрывает слушающее соединение и построенный им пул и возвращает ClosedSqlxBroker. Он считает блокировки, снятые с возвратом соединения в пул, и соединения, закрытые вместо этого.

Завершение работы само не возвращает захваченные строки. Каждая форма возвращает их по-своему: форма блокировки строк - когда её транзакция откатывается или её соединение закрывается, форма аренды - когда отброшенная доставка освобождает аренду или аренда истекает, форма рекомендательной блокировки - когда снимается её блокировка. В сервисе рантайм дожидается обработчиков, прежде чем завершить работу брокера, а обработчик, прерванный по тайм-ауту этого ожидания, отбрасывает свою доставку.