Cluster · разработчикам

Обмен юзерботов

Макет запроса:

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    config = {
        "peer": {"type": "string", "required": True},
        "module": {"type": "string", "required": True},
    }

    @loader.command("exchange", access="owner")
    async def exchange(self, ctx: Event) -> None:
        request = await app.exchange.request(
            await app.config.get("peer"),
            await app.config.get("module"),
            "main",
            {"type": "data", "key": "weather"},
            fields={"temperature": {"type": "number", "required": True}},
            direction="receive",
            mode="permanent",
        )
        await ctx.reply(request["id"])

В конфигураторе указываются ID другого юзербота и его модуля. Пример требует exchange.request, config.read, telegram.read, telegram.send. Экземпляр источника main должен хранить в weather объект с числовым temperature. Доступ к самому ресурсу проверяется только после согласия другой стороны.

Обмен работает внутри одной службы Cluster. По договору передаются данные, текст, файлы или события. Договор не открывает чужую базу целиком, не разрешает запускать команды и не даёт доступ к Telegram другого аккаунта.

Договор

await app.exchange.request(peer, module, slot, resource, fields=None,
                           mode="permanent", direction="receive", group=None)

• peer: str: постоянный ID другого юзербота.
• module: str, slot: str: ID модуля и имя его экземпляра у другой стороны.
• resource: dict[str, str]: ровно два поля, type и key.
• fields: dict | None: описание полей данных; формат как у конфигуратора.
• mode: str: permanent, once или group.
• direction: str: receive или send.
• group: str | None: общая метка договоров.
• Результат: словарь созданного договора.

Все методы app.exchange требуют exchange.request. Со стороны инициатора выбирается текущий экземпляр. Овнер через сайт также может использовать раздел ядра core/main.

receive получает ресурс другого юзербота, send передаёт ресурс текущего экземпляра. Инициатором может быть как источник, так и получатель. Согласие всегда даёт другая сторона.

permanent допускает повторные передачи. once допускает одну принятую передачу. group помечает несколько договоров общей меткой group: каждый остаётся отдельным направленным договором с отдельным согласием. Метка не создаёт автоматическую рассылку всем участникам.

Основные поля результата:

Ядро также хранит привязки установок и учёт квоты. Договор связан с профилем юзербота и конкретной установкой экземпляра. Смена Telegram-аккаунта договор сохраняет. Переустановка под тем же именем требует нового договора.

Ресурсы

key до 512 символов. Для file это допустимый ID файла. В fields до 64 полей, используются типы и ограничения из MODULES.md. Описание договора со служебным запасом должно помещаться в половину предела сообщения ядра, исходно 512 KiB.

Для data и text ядро получает текущее значение из раздела источника. Строка text до 16384 символов. Нужны текущие права источника на получение данных; для файла нужны его права files.read и доступ к самому файлу. Договор эти права не расширяет.

event содержит данные автора. Название потока не доказывает, что значения получены из Telegram или другого внешнего источника. Клиенты, файловые дескрипторы, ключи Telegram и выражения для выполнения не передаются: API принимает JSON и ID.

Согласие

Новый запрос ожидает согласия (pending). На юзербота разрешено до 20 запросов в минуту и до 100 ожидающих согласия. Овнер другой стороны принимает или отклоняет запрос на сайте через exchange/approve. Доверка может инициировать его из доступного ей модуля, но не одобрить.

До согласия наличие чужого ресурса не проверяется и не раскрывается. После согласия проверяются экземпляр, права, ресурс и подписки. Если ресурс отсутствует, договор сообщает о недоступности (unavailable).

Если участник остановлен, согласие сохраняется, а договор ждёт запуска участников (waiting). После запуска ядро повторяет проверку, договор разрешает обмен (active) или сообщает о недоступности (unavailable). Старая привязка не даёт доступ переустановленному экземпляру.

Состояния договора:

Изменение ресурса, направления или полей оформляется новым запросом с отдельным согласием.

Передача

await app.exchange.transfer(id, value=None)

id: str является ID активного договора. Для data, text и file значение получается из согласованного ресурса, value не требуется. Для event источник передаёт value: dict, до 16 KiB JSON. Только источник может отправить событие.

Результат: {"state": "delivered", "transfer": "...", "data": ...}. Для файла data содержит описание новой копии с ID получателя. Для остальных видов это переданные значения.

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.command("receive", access="owner")
    async def receive(self, ctx: Event) -> None:
        result = await app.exchange.transfer(ctx.args)
        await ctx.reply(result["state"])

Пример принимает ID уже согласованного договора в аргументе команды. Для потока источник вызывает await app.exchange.transfer(id, value={"temperature": 18}).

Ядро проверяет договор, обе установки, текущие права, работу участников, подписки и место. Остановленный участник сразу возвращает ошибку доступности. Очередь будущих передач на время остановки не создаётся.

Данные и текст сохраняются у получателя в ключе exchange:CONTRACT_ID. Файл копируется частями до 64 KiB с проверкой размера и SHA256. При ошибке частичная копия удаляется.

В истории различаются принятие передачи (accepted), доставка (delivered) и ошибка (failed). Доставка означает сохранение у получателя. Запуск его обработчика учитывается отдельно. Выполнение ровно один раз и автоматический повтор неизвестного внешнего результата не гарантируются.

Разовый договор (once) считается использованным (used) при принятии передачи. Последующая ошибка не возвращает согласие. Для повторной попытки при необходимости создаётся новый договор.

События

await app.exchange.events(id, after=0, limit=100)
await app.exchange.clear(id, before)

• id: str: ID договора с ресурсом event.
• after: int: последний полученный номер, от 0 до 2**53.
• limit: int: от 1 до 100.
• before: int: номер, до которого удаляются события включительно.

Форма результата для подсказок:

from typing import TypedDict

class ExchangeEvent(TypedDict):
    sequence: int
    time: float
    values: dict
    transfer: str

class EventsPage(TypedDict):
    events: list[ExchangeEvent]
    after: int

Это описание JSON, не экспорт SDK. Каждая отправка сохраняется отдельной записью получателя и не заменяет предыдущую. Размер порции также ограничен половиной предела сообщения ядра, поэтому событий может быть меньше limit.

Получатель передаёт полученный after в следующий запрос. Пустой список означает отсутствие новых сохранённых событий. Модуль сам выбирает способ получения и обработки, автоматического вызова обработчика нет.

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.command("events", access="owner")
    async def events(self, ctx: Event) -> None:
        id = ctx.args
        key = f"cursor:{id}"
        after = await app.db.get(key, default=0)
        page = await app.exchange.events(id, after=after)

        for entry in page["events"]:
            await app.db.set("last_event", entry["values"])
            await app.db.set(key, entry["sequence"])

        await ctx.reply(str(len(page["events"])))

Пример сохраняет курсор в базе и продолжает с него после перезапуска. Если действие с событием завершилось, а курсор записаться не успел, событие придёт снова. Для действий с последствиями нужна своя защита от повтора.

clear доступен получателю, возвращает {"removed": int} и освобождает место. Отправитель не может удалить полученные события. После отзыва старые записи остаются доступны получателю, новые передачи запрещены.

Отзыв

await app.exchange.revoke(id)

id: str, результатом является обновлённый договор. Отозвать может овнер своего участника или сам модуль. Доверка через действие модуля от своего имени это сделать не может.

Ядро упорядочивает принятие передачи и отзыв. Уже принятая передача может завершиться. Если отзыв произошёл первым, новая не начинается. Полученные файлы, значения и события при отзыве не удаляются.

Каждое событие является отдельной передачей. Долгая работа модуля не превращает весь поток в одну неостанавливаемую передачу.

Список и история

await app.exchange.get(id)
await app.exchange.list(after=None, limit=100)
await app.exchange.history(id)

id: str, after: str | None, limit: int от 1 до 100. get возвращает договор, list массив договоров по возрастанию ID, history массив записей передачи.

Модуль видит договоры своего экземпляра. Овнер на сайте видит договоры своего юзербота. Для следующей порции списка передаётся ID последнего договора в after. Окончание определяется пустой страницей. Большое описание сокращает порцию раньше limit, короткая непустая страница ещё не означает конец.

История содержит до 1000 записей с полями id, time, kind, size, state. Дополнительную копию содержимого она не хранит. Отдельной пагинации истории нет.

Квота

До согласия метаданные запроса занимают место только у инициатора. При одобрении их размер делится поровну между участниками. При отказе или отзыве доля другой стороны возвращается инициатору. Сам входящий запрос не расходует место адресата до согласия.

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

Перераспределение квоты между базами сохраняет отметку операции. После сбоя ядро завершает перенос учёта или возвращает прежний. Повтор восстановления не прибавляет резерв заново.

Хранение и проверки

Договоры и история хранятся в центральной базе. Данные участников находятся в их пользовательских базах и файлах. Модуль обращается к менеджеру обмена через закрытый канал ядра, чужой сокет или каталог ему не передаётся.

Обмен работает на одном узле. Между отдельными узлами потребуется защищённый транспорт и согласование владения данными; эта инфраструктура не реализована.

Тесты проверяют скрытие ресурса до согласия, одобрение при остановке, недоступность участника, смену аккаунта, переустановку, разовую передачу, отзыв, размер, квоту и контрольные суммы. Сценарий двух профилей передаёт файлы и данные через настоящее ядро с отдельными средами gVisor.