Исходный код, разобранный в статье, опубликован в этом репозитории

На текущий момент backend решения принято писать на микросервисах. Однако, в условиях отсутствия DevOps, микросервисы масштабироваться не будут, так как некому настраивать Envoy proxy: каждый микросервис работает в единственной реплике занимая целевой gRPC порт без проксирующей нагрузку прослойки.

Удешевляем микросервисы

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

Установка зависимостей

Ставить NGinx и Lua нужно на хост машину, так как мы заранее не знаем, какую ерись разработчики пропишут в docker networks: network_mode: host в этом случае не вариант, плюс хост машина не делает прокси поверх прокси

apt install nginx nginx-extras
sudo add-apt-repository ppa:ondrej/nginx
sudo apt update
apt install libnginx-mod-http-lua

sudo sysctl net.ipv4.ip_unprivileged_port_start=0
sudo tee -a /etc/sysctl.conf <<< "net.ipv4.ip_unprivileged_port_start=0"

Файл /etc/nginx/nginx.conf

Следующий NGinx конфиг будет ждать перезапуск backend 10 секунд. Если backend не ответил, для всех страниц возвращяется html с перезагрузкой страницы: fetch запросы упадут с ошибкой и фронт перезагрузится, тонкие клиенты на момент открытия веб страницы осуществят повтор попытки запуска приложения

load_module /usr/lib/nginx/modules/ndk_http_module.so;
load_module /usr/lib/nginx/modules/ngx_http_lua_module.so;

error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;

events {
    worker_connections 1024;
    use epoll;
    multi_accept on;
}

http {
    log_format main '$remote_addr - $remote_user [$time_local] "$request" '
                    '$status $body_bytes_sent "$http_referer" '
                    '"$http_user_agent"';

    access_log /var/log/nginx/access.log main;

    map $http_upgrade $connection_upgrade {
        default upgrade;
        ''      close;
    }

    server {
        listen 50050;
        server_name localhost;

        error_page 502 @wait_for_upstream;

        location / {
            proxy_pass http://127.0.0.1:8081$is_args$args;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection $connection_upgrade;
            proxy_set_header Host "localhost";
            proxy_set_header X-Real-IP "";
            proxy_set_header X-Forwarded-For "";
            proxy_set_header X-Forwarded-Proto "";
            proxy_hide_header Via;
            proxy_hide_header Server;

            proxy_next_upstream off;
            proxy_intercept_errors on;
            proxy_buffering off;
        }

        location @wait_for_upstream {
            content_by_lua_block {
                local function is_healthy()
                    local res = ngx.location.capture("/health_check", {
                        method = ngx.HTTP_GET,
                        args = "target=127.0.0.1:8081"
                    })
                    return res.status and res.status < 400
                end

                ngx.sleep(10)

                if not is_healthy() then
                    ngx.status = 200
                    ngx.header["Content-Type"] = "text/html"
                    ngx.say("<p>Updating...</p><script>setTimeout(() => { window.location.reload() }, 10_000)</script>")
                    ngx.exit(ngx.HTTP_OK)
                else
                    ngx.exec("@wait_for_upstream_proxy")
                end
            }
        }

        location @wait_for_upstream_proxy {
            proxy_pass http://127.0.0.1:8081$is_args$args;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection $connection_upgrade;
            proxy_set_header Host "localhost";
            proxy_set_header X-Real-IP "";
            proxy_set_header X-Forwarded-For "";
            proxy_set_header X-Forwarded-Proto "";
            proxy_hide_header Via;
            proxy_hide_header Server;

            proxy_next_upstream off;
            proxy_intercept_errors on;
            error_page 502 503 504 = @wait_for_upstream;
            proxy_buffering off;
        }
    }
}

Шина событий с бесшовным перезапуском реплик

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

В Linux несколько процессов могут обслуживать один и тот же порт. В отличие от nginx_upstream, нет очереди, которая тратит время на ожидание ответа от мертвого процесса. Дополнительно, когда event loop реплики заблокирован, управление возьмет на себя следующая реплика. Очень полезно, когда устройства интернета вещей отправляют медиа файлы в формате base64

Стандартным способом взаимодействия микросервисов в Linux является не сеть, а файлы. Это быстрее, так как файл можно монтировать в оперативную память. Например, такая шина событий используется в GETH - ноде блокчейна Ethereum, для соединения GUI с кошельком

ecosystem.config.js

На каждую реплику назначается условный порт. Он используется как идентификатор реплики для перезапуска, если процесс упадет. Внутри веб с Прописав больше портов через запятую, будут запущены дополнительные реплики

const path = require("path");
const os = require("os");
const dotenv = require("dotenv");

const readConfig = () => dotenv.parse("./.env");

const getPath = (unixPath) => {
  return path.resolve(unixPath.replace("~", os.homedir()));
};

const apps = [
  {
    name: "bun-ws-1",
    exec_mode: "fork",
    instances: "1",
    autorestart: true,
    cron_restart: "0 0 * * *",
    max_memory_restart: "1250M",
    script: "./packages/backend/src/index.ts",
    interpreter: getPath("~/.bun/bin/bun"),
    env: readConfig(),
    args: ["--bootstrap=8081,8082,8083,8084,8085"],
    out_file: "./logs/pm2/bun-ws-1-out.log",
    error_file: "./logs/pm2/bun-ws-1-error.log",
    log_date_format: "YYYY-MM-DD HH:mm:ss",
    merge_logs: true,
  },
];

module.exports = {
  apps,
};

Отправка данных

Для обмена событиями между репликами используется bootstrapService. Так как технически websocket соединение это не заканчивающийся HTTP запрос, он подключается только к одной реплике и не шарится между процессами по умолчанию.

import { errorData, getErrorMessage } from "functools-kit";
import type { NotifyRequest } from "../model/NotifyRequest.model";
import { app } from "../config/app";
import { ioc } from "../lib";

const { port } = ioc.bootstrapService.getArgs();

app.post("/api/v1/notify", async (ctx) => {
  const request = await ctx.req.json<NotifyRequest>();
  ioc.bootstrapService.broadcast("broadcast", {
    clientId: request.clientId,
    requestId: request.requestId,
    port,
  });
});

Отправка сообщения осуществляется через метод broadcast. Подписка на события в реплике осуществляется через метод listen

  ioc.bootstrapService.listen(({ topic, data }) => {
    if (server.subscriberCount(topic)) {
      console.log("Publishing", { topic, data });
      server.publish(topic, JSON.stringify(data));
    }
  });

Масштабируемый WebSocket

У bun есть pubsub вещание для ws соединений. Это когда вместо ссылки на объект используется строковый идентификатор

Таким образом, реплика может отправить сообщение на ws коннект другой реплики

app.get(
  "/api/v1/listen",
  upgradeWebSocket((c) => {
    return {
      onOpen: (_, ws) => {
        const bunWs = ws.raw!;
        bunWs.subscribe("broadcast");
        ws.send(JSON.stringify({ port }))
      },
      onClose: (_, ws) => {
        const bunWs = ws.raw!;
        bunWs.unsubscribe("broadcast");
      },
    };
  })
);

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