Исходящий HTTP, потоки WebSocket и электронная почта

01Исходящий HTTP

Приложение, которое обращается к другому серверу (вебхук, входящий вебхук Slack, сторонний API), использует Cybertrain::HttpClient. Он не входит в require "cybertrain"; подключайте его там, где он используется, а также слой TLS, если какой-либо URL — https:

require "json"
require "cybertrain/http_client"
require "cybertrain/net/tls"   # только для https://; компонуется с системной libssl

client = Cybertrain::HttpClient.new(timeout: 5, deadline: 15, max_body: 1 << 20)
begin
  res = client.post(ENV["SLACK_WEBHOOK_URL"], JSON.generate({ "text" => "Deployed" }),
                    content_type: "application/json")
  Cybertrain.logger.warn("slack answered #{res.status}") unless res.success?
rescue Cybertrain::HttpClient::Error => e
  Cybertrain.logger.warn("slack: #{e.message}")   # называет host:port, но никогда путь
end

res = client.get("https://api.example.com/v1/status", headers: { "Authorization" => "Bearer #{token}" })
res.status              # 200
res.header("ETag")      # имена в любом регистре; в res.headers они в нижнем регистре
res.body                # байты, с удалённым chunked-кодированием

Зачем он нужен: Net::HTTP из Spinel выполняет рукопожатие TLS и каждое чтение после первого байта блокирующими вызовами C, и каждый из них удерживает рабочий поток ОС — а при SPINEL_WORKERS=1 по умолчанию это весь веб-сервер — столько, сколько захочет зависший эндпоинт. HttpClient никогда не блокирует рабочий поток на сети:

• Каждое ожидание ограничено. Подключение (неблокирующий connect; TCPSocket.new из Spinel игнорирует connect_timeout:), каждый шаг рукопожатия TLS, каждое чтение и каждая запись ждут не более timeout секунд (по умолчанию 5), а весь запрос — не более deadline секунд (по умолчанию 15). При превышении любого из них — HttpClient::TimeoutError. Чтение идёт через байтовый буфер, наполняемый read_nonblock, поэтому уже полученные байты никогда не ожидаются.
• Разрешение имён блокирует. У getaddrinfo в среде выполнения нет неблокирующей формы: поток, выполняющий запрос, ждёт DNS (обычно миллисекунды, но неработающий резолвер может занять системный тайм-аут, около 5 с на попытку). Крайний срок учитывает это время, но не может его прервать.
• Непубличные адреса назначения отвергаются до того, как что-либо отправлено: loopback (127.0.0.0/8, ::1), частные (10/8, 172.16/12, 192.168/16, fc00::/7), link-local (169.254/16, куда входят облачные сервисы метаданных, fe80::/10), неопределённые (0.0.0.0/8, ::), multicast, broadcast, CGNAT (100.64/10), диапазоны для документации и бенчмарков, а также их формы IPv4-mapped, NAT64 и 6to4 (Cybertrain::Net::Address.blocked_reason). Имя хоста разрешается один раз, и проверяются и набираются именно те адреса, в которые оно разрешилось (повторно ничего не ищется), поэтому имя, разрешающееся в частный адрес (запись в /etc/hosts, DNS rebinding), отвергается так же, как литерал; сертификат TLS всё равно проверяется по имени. Проверка идёт по каждому адресу: отвергнутый адрес пропускается, и запрос завершается сбоем, только когда отвергнуты все адреса. HttpClient::BlockedAddressError сообщает причину. Чтобы обратиться к сервису в собственной сети или к локальному серверу в тесте, задайте c.http_allow_private = true в config/app.rb (или CYBERTRAIN_HTTP_ALLOW_PRIVATE=1) либо передайте allow_private: true в HttpClient.new; оставляйте это выключенным всякий раз, когда URL может прийти от пользователя.
• Перебираются все разрешённые адреса в порядке, который даёт резолвер, IPv6 и IPv4 одинаково: когда один отказывает, недоступен, сбрасывает соединение или не отвечает в пределах timeout, набирается следующий, все — в пределах deadline, а когда не удаётся ни один, возбуждается ошибка последнего. Именно благодаря этому localhost работает на macOS, где он разрешается сначала в ::1, а сервер, слушающий на 127.0.0.1, такое соединение отвергает. Cybertrain::Mailer подключается так же.
• TLS берётся из пакета openssl: VERIFY_PEER по хранилищу доверенных сертификатов операционной системы (SSL_CERT_FILE / SSL_CERT_DIR указывают на другое; у пакета нет X509::Store для собственного), проверка имени хоста, SNI, TLS 1.2 или новее. URL https нужно имя хоста: IP-адрес не совпадает с именем в сертификате. Без require "cybertrain/net/tls" запрос https возбуждает HttpClient::TLSError, который говорит об этом; один cybertrain/http_client никогда не компонуется с libssl.
• Клиент для уведомлений, а не браузер: HTTP/1.1 с Connection: close, одно соединение на запрос, без keep-alive, без редиректов (3xx возвращается как есть), без прокси, без cookie, без сжатия. Ответы могут быть с Content-Length, chunked или читаться до закрытия; заголовки ограничены 64 КиБ и 100 строками, тело — max_body (по умолчанию 1 МиБ), сверх которого — HttpClient::TooLargeError. Ответ, который не является HTTP/1.x или обрывается раньше времени, — это HttpClient::BadResponseError; отвергнутое или сброшенное соединение либо имя, которое не разрешается, — HttpClient::RefusedError. Соединение, которое сбрасывается или закрывается во время рукопожатия TLS (сервер, не говорящий на TLS), — это HttpClient::TLSError («TLS handshake failed») на любой платформе, как и рукопожатие, которое отвергает OpenSSL. Все они — HttpClient::Error.
• Ошибки никогда не содержат путь или строку запроса, в которых URL вебхука несёт свой секрет: сообщения называют только хост и порт. Некорректный URL или заголовок с CR/LF — это ArgumentError, который тоже его не повторяет. Host, Connection, Content-Length и Transfer-Encoding принадлежат клиенту; User-Agent по умолчанию — cybertrain/<version>.

Запрос выполняется в вызывающем потоке — зелёном потоке, как и обслуживаемый запрос, — поэтому, пока он ждёт, другие соединения продолжают обслуживаться. Слой сокетов (cybertrain/net/socket.rb: ограниченный connect, буферизованное чтение, ограниченная запись) отделён от слоя HTTP, чтобы им могли пользоваться другие протоколы. Проектное решение: docs/design.md D24.

02Потребление потоков WebSocket

Приложение, которое слушает внешний поток (биржевые тики, телеметрию устройств, поток событий), пока работает сервер, подписывается на него через Cybertrain::WebSocket.subscribe. Как и HTTP-клиент, он не входит в require "cybertrain"; подключайте его там, где пишется подписка, а для wss:// — также слой TLS:

# config/app.rb (или файл, который bin/<app>.rb подключает до Cybertrain::Main.run)
require "json"
require "cybertrain/websocket"
require "cybertrain/net/tls"   # только для wss://; компонуется с системной libssl

Cybertrain::WebSocket.subscribe("trades", ENV["FEED_URL"],
    headers: { "Authorization" => "Bearer #{ENV["FEED_TOKEN"]}" },
    stale_after: 30)
  .on_open    { |conn| conn.send_text(JSON.generate({ "op" => "subscribe", "channel" => "trades" })) }
  .on_message { |conn, text| Cybertrain::DB.with { |c| c.execute("INSERT INTO trades(raw) VALUES (?)", [text]) } }

Подключение cybertrain/websocket регистрирует потоки в Cybertrain::Background, который Application#serve запускает после планировщика и останавливает при завершении работы, так что вызывать ничего не нужно. Каждая подписка получает один поток выполнения. Он подключается, выполняет on_open (отправляйте там своё сообщение о подписке: оно выполняется снова после каждого переподключения), а затем передаёт каждое сообщение в on_message вместе с Connection (send_text, send_binary, close и last_opcode — :text или :binary). Блоки получают свои аргументы и не могут использовать self (в Spinel нет instance_eval).

• Когда это выполняется. Потоки запускаются, когда бинарный файл обслуживает запросы, как и планировщик (Фоновая работа); команды migrate, db ..., Main.command и Cybertrain::Test::Client никогда их не запускают. С CYBERTRAIN_WORKER=0 (config.worker = false) они не работают в этом процессе (они просто не запускаются; единственное уведомление — строка планировщика «run elsewhere», а WebSocket.start!, вызванный напрямую, пишет в лог свою): когда запросы обслуживают несколько процессов, отключите это во всех, кроме одного, иначе каждый получит весь поток. subscribe нужно вызвать до запуска сервера; добавлять или удалять поток во время работы нельзя. Имя уникально (ArgumentError), а поток без on_message срывает запуск до того, как занят порт (error: WebSocket subscription "trades" has no on_message, код выхода 1).
• Параметры. headers: (дополнительные заголовки рукопожатия), timeout: (5 секунд на каждое ожидание при подключении и на каждую запись), max_message: (1 МиБ; более крупное сообщение закрывает соединение с кодом 1009), ping_interval: (15) и stale_after: (45), а также allow_private: (как у HttpClient, ниже).
• Ожидание и переподключение. После любого сбоя поток ждёт min(60, 2 ** (attempt - 1)) секунд с разбросом 20 % в обе стороны: 1, 2, 4, ... 60. Соединение, продержавшееся 60 секунд, начинает снова с 1. Каждое ожидание ограничено. WebSocket.stop! возвращается примерно через 2 секунды, что бы ни делали серверы: он ждёт не более 2 секунд все потоки вместе, подключённый поток отправляет close (1000) и ждёт ответа сервера не более 1 секунды, и после возврата connected? каждого потока ложно (поток, который всё ещё закрывается, завершается сам).
• Тишина. Поток может замолчать, так и не закрыв соединение, поэтому поток выполнения отправляет пинг после ping_interval секунд без единого фрейма (и затем раз в интервал), а когда за stale_after секунд не пришло ни одного фрейма, включая pong, закрывает соединение и переподключается. stale_after должен быть больше ping_interval, который должен быть не меньше 1 (ArgumentError от subscribe).
• Ошибки в ваших блоках. Исключение в on_message пишется в лог, и соединение остаётся. Исключение в on_open пишется в лог, соединение закрывается и повторяется, поскольку от потока, который не подписался, нет пользы.

• Что пишется в лог.

  Когда	Уровень	Строка
  запуск	INFO	WebSocket started: trades (wss://feed.example.com:443)
  подключение	INFO	WebSocket "trades" connected to feed.example.com:443
  отключение	WARN	WebSocket "trades" disconnected: Cybertrain::WebSocket::ClosedError: feed.example.com:443: closed by the server (1001) (retry in 2s)
  тишина	WARN	WebSocket "trades" silent for 45s; reconnecting
  блок возбудил исключение	ERROR	WebSocket "trades" on_message failed: KeyError: ... (или on_open)

  Путь и строка запроса никогда не пишутся в лог и не попадают в сообщения об ошибках: в них часто содержится токен.

• Ошибки. Все они — Cybertrain::WebSocket::Error, и в сообщении только хост и порт. После рукопожатия потерянное соединение или закрытие сервером — это ClosedError (close_code, close_reason; сброс или конец без close-фрейма — 1006); запись, которая превышает тайм-аут или не удаётся в TLS, — по-прежнему TimeoutError или TLSError. Во время подключения это RefusedError, TimeoutError, TLSError, BlockedAddressError (непубличный адрес, по умолчанию отвергаемый ровно так же, как в Исходящий HTTP, с тем же allow_private: и настройкой http_allow_private) или HandshakeError (не 101, неверный Sec-WebSocket-Accept или сервер, ответивший с расширением или подпротоколом). ProtocolError (сервер нарушил формат фреймов; клиент отправляет 1002) и TooLargeError могут возникнуть в любой момент.
• Состояние. Cybertrain::WebSocket.feed_status("trades") возвращает Status: connected?, last_message_at (секунды эпохи, 0 до первого сообщения) и reconnects. Это не входит в GET /up; используйте это в собственной проверке. Все три читаются одно за другим, а не как единый снимок, и connected? становится истинным, когда соединение установлено, до того как выполнился on_open.
• Делайте on_message коротким. Поток выполнения ждёт через wait_readable или sleep, поэтому зависший поток данных не удерживает рабочий поток ОС, пока ждёт. Удерживают его две вещи: каждая попытка подключения разрешает имя хоста блокирующим getaddrinfo (см. «Разрешение имён блокирует» в Исходящий HTTP; при SPINEL_WORKERS=1 неработающий резолвер задерживает каждый запрос на тайм-аут резолвера, по разу на попытку на поток, а IP-адрес в URL позволяет обойтись без поиска) и вызов SQLite внутри on_message до своего возврата, как в задаче по расписанию (Фоновая работа): при SPINEL_WORKERS=1 долгая запись задерживает каждый запрос. Для нагруженного потока вставляйте одну небольшую строку или собирайте сообщения в памяти и записывайте их пакетами из задачи Scheduler.every.
• Не реализовано. Расширения (permessage-deflate) и подпротоколы не согласовываются. Нет поддержки прокси, редиректов и cookie, нет автоматического сохранения сообщений и нет сервера WebSocket (ActionCable). Текстовые фреймы не проверяются на допустимый UTF-8 (отступление от RFC 6455, раздел 8.1): байты передаются дальше в том виде, в каком пришли. Connection действителен только во время колбэка; отправка из запроса или задачи не поддерживается.

Для одного соединения без цикла переподключения (скрипт, однократное чтение) используйте синхронный клиент:

require "cybertrain/websocket"
require "cybertrain/net/tls"   # только для wss://

client = Cybertrain::WebSocket::Client.connect("wss://feed.example.com/v1", timeout: 5)
client.send_text("hello")
msg = client.receive(10)      # ждёт не более 10 с; nil, если ничего не пришло
msg.ws_opcode                 # :text или :binary
msg.ws_data                   # байты
client.close

Проектное решение: docs/design.md D33.