Merge pull request 'feat(parser): авто-переподключение к SHiP с backoff (1.3.0)' (#12) from dev into main

ReconnectSupervisor был написан, протестирован и экспортнут, но НИГДЕ не
подключён к циклу чтения. Parser.start() имел голый for-await без try/catch:
падение ноды → ShipConnectionError наверх → темп переподключений целиком
зависел от внешнего супервизора процесса. Мёртвая нода = рестарт без пауз =
долбёжка ноды без ограничителей.

Что сделано:
- Цикл чтения вынесен в runStreamLoop (core/streamLoop.ts) с инъекцией
  зависимостей — чтобы покрыть reconnect/resume юнит-тестами на фейках, без
  реального Redis и SHiP.
- runStreamLoop оборачивает (пере)подключение в ReconnectSupervisor:
  backoff 1→2→5→15→60с, после maxAttempts без прогресса — process.exit(1)
  (оркестратор перезапустит штатно).
- Позиция возобновления перечитывается из Redis на КАЖДОМ подключении
  (sync-hash пишется после каждого блока) — продолжаем ровно с последнего
  записанного блока, без потерь и дублей.
- ReconnectSupervisor.run теперь передаёт fn callback resetBackoff: каждый
  обработанный блок сбрасывает счётчик попыток, чтобы долгая стабильная сессия
  после редкого разрыва не копила attempt до выхода.
- Штатная остановка (stop закрывает ws) ловится и не вызывает лишний backoff.
- Parser получил структурный логгер (opts.logger) для onAttempt/onGiveUp.

Тесты: +4 runStreamLoop (reconnect+resume, stop без reconnect, give-up на
мёртвой ноде, resetBackoff на прогрессе), +4 ReconnectSupervisor (resetBackoff).
Всего 229 unit. README: секция «Авто-переподключение к SHiP».

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

README.md

@@ -49,6 +49,11 @@ abiFallback: rpc-current
 xtrim:
   enabled: true
   intervalMs: 60000
+backpressure:
+  enabled: true       # пауза чтения SHiP когда стрим переполнен (защита Redis RAM)
+  highWater: 100000   # порог паузы — число событий в стриме
+  lowWater: 50000     # порог возобновления (должен быть < highWater)
+  pollMs: 200         # период опроса XLEN во время паузы, мс
 logger:
   level: info
   pretty: false
@@ -185,6 +190,11 @@ abiFallback: rpc-current
 xtrim:
   enabled: true
   intervalMs: 60000
+backpressure:
+  enabled: true       # пауза чтения SHiP когда стрим переполнен (защита Redis RAM)
+  highWater: 100000   # порог паузы — число событий в стриме
+  lowWater: 50000     # порог возобновления (должен быть < highWater)
+  pollMs: 200         # период опроса XLEN во время паузы, мс
 reconnect:
   maxAttempts: 10
   backoffSeconds: [1, 2, 5, 10, 30, 60, 120, 300, 600, 1800]
@@ -224,7 +234,8 @@ docker run --rm --network host \
                         │  • Workers   │            │              │
                         │  • ForkDet.  │            │  Streams +   │
                         │  • XtrimSup. │            │  Hashes +    │
-                        └──────────────┘            │  ZSets       │
+                        │  • Backpr.   │            │  ZSets       │
+                        └──────────────┘            │              │
                                                     │              │
                         ┌──────────────┐            │              │
                         │ ParserClient │◀─XREADGR──│              │
@@ -240,6 +251,57 @@ docker run --rm --network host \
 - [Redis key taxonomy](docs/redis-key-taxonomy.md)
 - [Disaster recovery / fork scenarios](docs/disaster-recovery.md)
 
+## Backpressure: защита памяти Redis на репарсинге
+
+Parser читает блоки из SHiP и пишет события в Redis-стрим. На длинном репарсинге (genesis → head, миллионы блоков) писатель легко обгоняет потребителя: события копятся в **памяти Redis** быстрее, чем `ParserClient` их вычитывает. RAM самого парсера ограничена окном SHiP (`max_messages_in_flight` + pull-based чтение с ack после обработки), а вот стрим в Redis может расти неограниченно — вплоть до OOM. `XtrimSupervisor` тут не спасает: он по дизайну не режет непрочитанные события.
+
+`BackpressureGate` вводит **окно поглощения с гистерезисом**. Перед обработкой каждого блока парсер смотрит backlog = `XLEN` стрима:
+
+- backlog ≥ **highWater** → чтение SHiP встаёт на **паузу** (блок не подтверждается, SHiP сам перестаёт слать после своего in-flight окна);
+- во время паузы парсер опрашивает `XLEN` каждые `pollMs` и ждёт, пока потребитель сольёт backlog ≤ **lowWater**;
+- слилось → чтение **возобновляется**.
+
+Ничего не теряется — события остаются в стриме, парсер лишь перестаёт доливать. Если потребитель вообще не подключён, backlog дорастает до `highWater` и писатель замирает, удерживая Redis в пределах ~`highWater` событий.
+
+**Почему не дёргается.** Разрыв high/low (по умолчанию 100k / 50k) — это антидребезг (гистерезис). Один цикл «пауза ↔ работа» перемалывает 50k событий, поэтому переключения происходят редко, а не на каждом блоке. В steady-state (потребитель у головы цепи) backlog почти нулевой → пауза не включается вообще, оверхед = один `XLEN` на блок (O(1)).
+
+```
+backlog
+100k |      /|        /|        /|       <- highWater -> ПАУЗА
+     |     / |       / |       / |
+ 50k |    /  |______/  |______/  |___    <- lowWater  -> ПУСК
+     |   /      (пауза)   (пауза)
+   0 +--+----------------------------> время
+        пишем  ждём  пишем  ждём  пишем
+```
+
+Конфиг (включён по умолчанию):
+
+```yaml
+backpressure:
+  enabled: true       # выключить можно enabled: false
+  highWater: 100000   # порог паузы — число событий в стриме
+  lowWater: 50000     # порог возобновления (< highWater)
+  pollMs: 200         # период опроса XLEN во время паузы, мс
+```
+
+Рекомендация по инфраструктуре: на Redis выставить `maxmemory` + `maxmemory-policy noeviction` — тогда при достижении лимита `XADD` вернёт ошибку (парсер её обработает), а не сработает OOM-killer.
+
+## Авто-переподключение к SHiP
+
+Если SHiP-нода падает или рвёт соединение, парсер **не пытается переподключиться мгновенно в цикле** — это превратило бы упавшую ноду в мишень для долбёжки без пауз. Вместо этого `ReconnectSupervisor` переподключается с **экспоненциальным backoff** (по умолчанию `1 → 2 → 5 → 15 → 60` секунд, дальше держит последнее значение):
+
+- разрыв соединения → пауза по таблице `backoffSeconds`, затем новая попытка;
+- успешная попытка перечитывает позицию возобновления из Redis (`sync`-hash пишется после каждого блока) и продолжает **ровно с последнего записанного блока** — без потерь и дублей;
+- каждый обработанный блок сбрасывает счётчик попыток: долгая стабильная сессия после редкого разрыва не копит попытки;
+- `maxAttempts` неудач подряд (нода действительно мертва, прогресса нет) → парсер завершается с кодом 1, чтобы оркестратор (Docker/systemd/k8s) перезапустил процесс штатно.
+
+```yaml
+reconnect:
+  maxAttempts: 10
+  backoffSeconds: [1, 2, 5, 10, 30, 60, 120, 300, 600, 1800]
+```
+
 ## Известные ограничения
 
 ### Схлопывание дельт create+remove внутри одного блока
@@ -298,8 +360,8 @@ pnpm --filter @coopenomics/parser2 test:integration   # нужен Docker для
 ```
 
 **Текущее покрытие:**
-- `@coopenomics/parser2`: 81% statements / 74% functions (205 unit-тестов)
-- `@coopenomics/coopos-ship-reader`: 81% statements / 86% functions (51 unit-тест)
+- `@coopenomics/parser2`: 229 unit-тестов + integration (xtrim/backpressure на реальном Redis)
+- `@coopenomics/coopos-ship-reader`: 54 unit-теста
 
 ### Бенчмарк
 

packages/parser2/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@coopenomics/parser2",
-  "version": "1.1.3",
+  "version": "1.3.0",
   "description": "Universal EOSIO/Antelope SHiP-to-Redis blockchain indexer (parser)",
   "license": "MIT",
   "author": "Coopenomics contributors",

packages/parser2/src/adapters/IoRedisStore.ts

@@ -43,6 +43,7 @@ interface IRedisClient {
   xrange(key: string, start: string, end: string, count: string, countVal: number): Promise<Array<[string, string[]]>>
   xrevrange(key: string, end: string, start: string, count: string, countVal: number): Promise<Array<[string, string[]]>>
   xlen(key: string): Promise<number>
+  del(...keys: string[]): Promise<number>
   xdel(key: string, ...ids: string[]): Promise<number>
   xack(stream: string, group: string, id: string): Promise<number>
   zadd(key: string, score: number, member: string): Promise<number>

packages/parser2/src/config/index.ts

@@ -40,6 +40,12 @@ export interface ParserOptions {
   abiFallback?: 'rpc-current' | 'fail'
   /** XtrimSupervisor: интервал проверки и включение/отключение автообрезки стрима. */
   xtrim?: { intervalMs?: number; enabled?: boolean }
+  /**
+   * Backpressure-окно: пауза чтения SHiP когда backlog стрима (XLEN) достигает
+   * highWater, возобновление при сливе до lowWater. Защита Redis RAM на
+   * репарсинге, когда consumer медленнее писателя. По умолчанию включено.
+   */
+  backpressure?: { enabled?: boolean; highWater?: number; lowWater?: number; pollMs?: number }
   /** ReconnectSupervisor: максимум попыток и backoff-таблица в секундах. */
   reconnect?: { maxAttempts?: number; backoffSeconds?: number[] }
   /** Десериализатор ABI-данных. Единственный вариант — 'wharfkit'. */

packages/parser2/src/config/schema.ts

@@ -82,6 +82,26 @@ export const configSchema = {
       },
       additionalProperties: false,
     },
+    /**
+     * Backpressure: окно поглощения между писателем (SHiP) и consumer'ом.
+     * Когда backlog стрима (XLEN) ≥ highWater — чтение блоков встаёт на паузу,
+     * пока consumer не сольёт backlog ≤ lowWater. Бережёт Redis RAM на длинном
+     * репарсинге. Ничего не теряет — события остаются в стриме.
+     */
+    backpressure: {
+      type: 'object',
+      properties: {
+        /** Включить/отключить. По умолчанию true. */
+        enabled: { type: 'boolean', default: true },
+        /** Верхняя граница backlog (XLEN) — пауза чтения. По умолчанию 100000. */
+        highWater: { type: 'number', default: 100000 },
+        /** Нижняя граница — возобновление. Должна быть < highWater. По умолчанию 50000. */
+        lowWater: { type: 'number', default: 50000 },
+        /** Интервал опроса XLEN во время паузы, мс. По умолчанию 200. */
+        pollMs: { type: 'number', default: 200 },
+      },
+      additionalProperties: false,
+    },
     /** ReconnectSupervisor: поведение при разрыве SHiP соединения. */
     reconnect: {
       type: 'object',

packages/parser2/src/core/BackpressureGate.ts

@@ -0,0 +1,100 @@
+/**
+ * Backpressure-окно между чтением блокчейна (писатель) и consumer'ом (читатель).
+ *
+ * Проблема: Parser читает блоки из SHiP и XADD'ит события в Redis-стрим быстрее,
+ * чем ParserClient (controller) их вычитывает. На длинном репарсинге (genesis →
+ * head, миллионы блоков) стрим растёт в ПАМЯТИ REDIS неограниченно → Redis OOM.
+ * RAM самого парсера ограничена SHiP-окном (max_messages_in_flight), а вот
+ * Redis-стрим — нет. XtrimSupervisor не спасает: он не режет непрочитанное.
+ *
+ * Решение: перед обработкой каждого блока проверяем backlog = XLEN стрима.
+ * Если backlog >= highWater — СТАВИМ чтение на паузу (не ack'аем блок, SHiP
+ * перестаёт слать) и ждём пока consumer сольёт backlog до lowWater. Потом
+ * продолжаем. Ничего не теряем: события лежат в стриме, мы лишь не доливаем.
+ *
+ * Если consumer вообще не подключён — backlog растёт до highWater и писатель
+ * замирает, держа Redis в пределах ~highWater событий.
+ *
+ * Почему XLEN, а не lag группы: XLEN — это ровно тот объём, что занят в Redis
+ * RAM (единственный защищаемый ресурс). Не зависит от числа/наличия групп и
+ * версии Redis (lag появился в 7.0). Консервативен: считает и непрочитанное,
+ * и ещё не обрезанное XtrimSupervisor'ом — пауза чуть раньше, это безопасно.
+ */
+
+import type { RedisStore } from '../ports/RedisStore.js'
+import { rootLogger, type Logger } from '../logger.js'
+
+const sleep = (ms: number): Promise<void> => new Promise(resolve => setTimeout(resolve, ms))
+
+export interface BackpressureGateOpts {
+  redis: RedisStore
+  /** Стрим, backlog которого ограничиваем (ce:parser2:<chainId>:events). */
+  stream: string
+  /** Достигли этого XLEN — пауза чтения SHiP. */
+  highWater: number
+  /** Возобновляем чтение когда backlog слился до этого XLEN. < highWater. */
+  lowWater: number
+  /** Интервал опроса XLEN во время паузы, мс. По умолчанию 200. */
+  pollMs?: number
+  /** Логгер (по умолчанию rootLogger). */
+  logger?: Logger
+}
+
+export class BackpressureGate {
+  private readonly redis: RedisStore
+  private readonly stream: string
+  private readonly highWater: number
+  private readonly lowWater: number
+  private readonly pollMs: number
+  private readonly log: Logger
+  private pausedNow = false
+
+  constructor(opts: BackpressureGateOpts) {
+    if (!Number.isFinite(opts.highWater) || opts.highWater <= 0) {
+      throw new Error(`BackpressureGate: highWater must be > 0, got ${opts.highWater}`)
+    }
+    if (!Number.isFinite(opts.lowWater) || opts.lowWater < 0 || opts.lowWater >= opts.highWater) {
+      throw new Error(`BackpressureGate: lowWater must be in [0, highWater), got ${opts.lowWater} (highWater=${opts.highWater})`)
+    }
+    this.redis = opts.redis
+    this.stream = opts.stream
+    this.highWater = opts.highWater
+    this.lowWater = opts.lowWater
+    this.pollMs = opts.pollMs ?? 200
+    this.log = (opts.logger ?? rootLogger).child({ component: 'BackpressureGate', stream: opts.stream })
+  }
+
+  /** true пока чтение блоков стоит на паузе (для метрик/health). */
+  get isPaused(): boolean {
+    return this.pausedNow
+  }
+
+  /**
+   * Блокирует выполнение пока backlog (XLEN стрима) >= highWater.
+   * Возвращается когда backlog опустился < lowWater ИЛИ shouldStop() стал true
+   * (graceful shutdown). Если backlog уже < highWater — возвращается сразу,
+   * не делая лишних RTT после первого XLEN.
+   *
+   * @param shouldStop предикат прерывания (обычно () => this.stopSignal)
+   */
+  async waitForCapacity(shouldStop: () => boolean = () => false): Promise<void> {
+    let backlog = await this.redis.xlen(this.stream)
+    if (backlog < this.highWater) return
+
+    this.pausedNow = true
+    this.log.warn(
+      { backlog, highWater: this.highWater, lowWater: this.lowWater },
+      'backpressure: чтение блоков на паузе — consumer отстаёт, ждём слива стрима',
+    )
+    try {
+      while (!shouldStop()) {
+        await sleep(this.pollMs)
+        backlog = await this.redis.xlen(this.stream)
+        if (backlog <= this.lowWater) break
+      }
+    } finally {
+      this.pausedNow = false
+    }
+    this.log.info({ backlog, lowWater: this.lowWater }, 'backpressure: backlog слит, чтение возобновлено')
+  }
+}

packages/parser2/src/core/Parser.ts

@@ -6,11 +6,16 @@ import { WorkerPool } from '../workers/WorkerPool.js'
 import { BlockProcessor } from './BlockProcessor.js'
 import type { XtrimSupervisorOpts } from './XtrimSupervisor.js'
 import { XtrimSupervisor } from './XtrimSupervisor.js'
+import { BackpressureGate } from './BackpressureGate.js'
+import { ReconnectSupervisor } from './ReconnectSupervisor.js'
+import { runStreamLoop } from './streamLoop.js'
 import { ForkDetector } from './ForkDetector.js'
 import { RedisKeys } from '../redis/keys.js'
 import { ChainIdMismatchError } from '../errors.js'
 import { AbiStore } from '../abi/AbiStore.js'
 import { AbiBootstrapper } from '../abi/AbiBootstrapper.js'
+import { createLogger } from '../logger.js'
+import type { Logger } from '../logger.js'
 import type { ParserEvent } from '../types.js'
 
 export class Parser {
@@ -20,6 +25,8 @@ export class Parser {
   private workerPool: WorkerPool | null = null
   private blockProcessor: BlockProcessor | null = null
   private xtrimSupervisor: XtrimSupervisor | null = null
+  private backpressureGate: BackpressureGate | null = null
+  private log: Logger | null = null
   private running = false
   private stopSignal = false
 
@@ -66,6 +73,12 @@ export class Parser {
       throw new ChainIdMismatchError(this.opts.chain.id, chainId)
     }
 
+    this.log = createLogger({
+      ...(this.opts.logger?.level !== undefined ? { level: this.opts.logger.level } : {}),
+      ...(this.opts.logger?.pretty !== undefined ? { pretty: this.opts.logger.pretty } : {}),
+      chain_id: chainId,
+    }).child({ component: 'Parser' })
+
     const abiFallback = this.opts.abiFallback ?? 'rpc-current'
     const abiStore = new AbiStore(this.redis)
     const abiBootstrapper = new AbiBootstrapper(this.chainClient, abiStore, { abiFallback })
@@ -81,14 +94,6 @@ export class Parser {
     const syncKey = RedisKeys.syncHash(chainId)
     const eventsStream = RedisKeys.eventsStream(chainId)
 
-    const lastBlockNum = await this.redis.hget(syncKey, 'block_num')
-    const lastBlockId = await this.redis.hget(syncKey, 'block_id')
-
-    const havePositions =
-      lastBlockNum && lastBlockId
-        ? [{ blockNum: Number(lastBlockNum), blockId: lastBlockId }]
-        : []
-
     const xtrimOpts: XtrimSupervisorOpts = {
       redis: this.redis,
       stream: eventsStream,
@@ -100,38 +105,54 @@ export class Parser {
       this.xtrimSupervisor.start()
     }
 
-    const streamOpts = {
-      startBlock: havePositions[0]?.blockNum ?? 0,
-      havePositions,
+    // Backpressure-окно: пауза чтения SHiP когда backlog стрима упирается в
+    // highWater, чтобы Redis не рос неограниченно на репарсинге. По умолчанию вкл.
+    if (this.opts.backpressure?.enabled !== false) {
+      const bp = this.opts.backpressure
+      this.backpressureGate = new BackpressureGate({
+        redis: this.redis,
+        stream: eventsStream,
+        highWater: bp?.highWater ?? 100_000,
+        lowWater: bp?.lowWater ?? 50_000,
+        ...(bp?.pollMs !== undefined ? { pollMs: bp.pollMs } : {}),
+      })
     }
 
     const irreversibleOnly = this.opts.irreversibleOnly ?? false
     const forkDetector = new ForkDetector(chainId)
 
-    for await (const block of this.chainClient.streamBlocks(streamOpts)) {
-      if (this.stopSignal) break
+    // ReconnectSupervisor оборачивает (пере)подключение к SHiP экспоненциальным
+    // backoff. Без него разрыв ноды бросал ShipConnectionError наверх из start(),
+    // и темп переподключений целиком зависел от внешнего супервизора процесса —
+    // мёртвая нода = рестарт без пауз = долбёжка. Теперь паузы 1→2→5→15→60с,
+    // а после maxAttempts подряд неудач — чистый выход (процесс рестартит оркестратор).
+    const reconnectCfg = this.opts.reconnect ?? {}
+    const supervisor = new ReconnectSupervisor({
+      ...(reconnectCfg.maxAttempts !== undefined ? { maxAttempts: reconnectCfg.maxAttempts } : {}),
+      ...(reconnectCfg.backoffSeconds !== undefined ? { backoffSeconds: reconnectCfg.backoffSeconds } : {}),
+      onAttempt: (attempt, delayMs) => {
+        this.log?.warn({ attempt, delayMs }, 'SHiP-соединение потеряно, переподключение через backoff')
+      },
+      onGiveUp: (attempts) => {
+        this.log?.error({ attempts }, 'SHiP-переподключение исчерпало все попытки, остановка парсера')
+        if (!this.opts.noSignalHandlers) process.exit(1)
+      },
+    })
 
-      if (irreversibleOnly && block.thisBlock.blockNum > block.lastIrreversible.blockNum) {
-        this.chainClient.ack(1)
-        continue
-      }
-
-      const forkEvent = forkDetector.check(block.thisBlock.blockNum, block.thisBlock.blockId)
-      const events: ParserEvent[] = await this.blockProcessor.process(block)
-      const toPublish: ParserEvent[] = forkEvent ? [forkEvent, ...events] : events
-
-      for (const event of toPublish) {
-        await this.redis.xadd(eventsStream, this.eventToFields(event))
-      }
-
-      await this.redis.hset(syncKey, {
-        block_num: String(block.thisBlock.blockNum),
-        block_id: block.thisBlock.blockId,
-        last_updated: new Date().toISOString(),
-      })
-
-      this.chainClient.ack(1)
-    }
+    await runStreamLoop({
+      chainClient: this.chainClient!,
+      redis: this.redis!,
+      blockProcessor: this.blockProcessor!,
+      backpressureGate: this.backpressureGate,
+      forkDetector,
+      supervisor,
+      syncKey,
+      eventsStream,
+      irreversibleOnly,
+      eventToFields: (event) => this.eventToFields(event),
+      isStopped: () => this.stopSignal,
+      log: this.log,
+    })
   }
 
   private eventToFields(event: ParserEvent): Record<string, string> {

packages/parser2/src/core/ReconnectSupervisor.ts

@@ -51,9 +51,15 @@ export class ReconnectSupervisor {
   /**
    * Запускает fn и повторяет при исключении с паузами.
    *
+   * fn получает callback resetBackoff: вызови его, когда fn сделала реальный
+   * прогресс (например успешно обработала блок на свежем соединении) — счётчик
+   * попыток сбросится в 0. Без сброса долгоживущее соединение, которое изредка
+   * рвётся, копило бы attempt до maxAttempts и завершало процесс, хотя между
+   * разрывами оно часами работало штатно.
+   *
    * Псевдокод:
    *   loop:
-   *     try: return await fn()
+   *     try: return await fn(resetBackoff)
    *     catch: attempt++
    *       if attempt >= maxAttempts: onGiveUp(); throw
    *       delay = backoffSeconds[min(attempt-1, len-1)] * 1000
@@ -61,11 +67,12 @@ export class ReconnectSupervisor {
    *
    * @returns Результат первого успешного вызова fn.
    */
-  async run<T>(fn: () => Promise<T>): Promise<T> {
+  async run<T>(fn: (resetBackoff: () => void) => Promise<T>): Promise<T> {
     let attempt = 0
+    const resetBackoff = (): void => { attempt = 0 }
     for (;;) {
       try {
-        return await fn()
+        return await fn(resetBackoff)
       } catch (err) {
         attempt++
         if (attempt >= this.maxAttempts) {

packages/parser2/src/core/streamLoop.ts

@@ -0,0 +1,129 @@
+/**
+ * Цикл чтения блоков SHiP с авто-переподключением.
+ *
+ * Вынесен из Parser.start() отдельной функцией с инъекцией зависимостей —
+ * чтобы покрыть reconnect/resume-логику юнит-тестами на фейках, без реального
+ * Redis и SHiP-ноды.
+ *
+ * Контракт:
+ *   - supervisor оборачивает одну попытку соединения экспоненциальным backoff;
+ *   - разрыв ноды → streamBlocks бросает → supervisor ждёт паузу и зовёт заново;
+ *   - позиция возобновления перечитывается из Redis на КАЖДОМ подключении
+ *     (sync-hash пишется после каждого блока) — продолжаем ровно с последнего
+ *     записанного блока, без потерь и дублей;
+ *   - resetBackoff() на каждом обработанном блоке: реальный прогресс возвращает
+ *     лесенку backoff в начало, чтобы долгая стабильная сессия после редкого
+ *     разрыва не копила попытки до выхода;
+ *   - штатная остановка (isStopped()) закрывает сокет извне → for-await бросает;
+ *     ловим и выходим без backoff и без retry.
+ */
+
+import type { ShipBlock, GetBlocksOptions } from '@coopenomics/coopos-ship-reader'
+import type { ReconnectSupervisor } from './ReconnectSupervisor.js'
+import type { BackpressureGate } from './BackpressureGate.js'
+import type { ForkDetector } from './ForkDetector.js'
+import type { Logger } from '../logger.js'
+import type { ParserEvent } from '../types.js'
+
+/** Узкий срез ChainClient, нужный циклу (упрощает фейки в тестах). */
+export interface StreamLoopChainClient {
+  connect(): Promise<unknown>
+  streamBlocks(opts: GetBlocksOptions): AsyncIterable<ShipBlock>
+  ack(n: number): void
+}
+
+/** Узкий срез RedisStore, нужный циклу. */
+export interface StreamLoopRedis {
+  hget(key: string, field: string): Promise<string | null>
+  hset(key: string, fields: Record<string, string>): Promise<unknown>
+  xadd(stream: string, fields: Record<string, string>): Promise<unknown>
+}
+
+/** Узкий срез BlockProcessor. */
+export interface StreamLoopBlockProcessor {
+  process(block: ShipBlock): Promise<ParserEvent[]>
+}
+
+export interface StreamLoopDeps {
+  chainClient: StreamLoopChainClient
+  redis: StreamLoopRedis
+  blockProcessor: StreamLoopBlockProcessor
+  backpressureGate: BackpressureGate | null
+  forkDetector: Pick<ForkDetector, 'check'>
+  supervisor: ReconnectSupervisor
+  syncKey: string
+  eventsStream: string
+  irreversibleOnly: boolean
+  eventToFields: (event: ParserEvent) => Record<string, string>
+  isStopped: () => boolean
+  log: Logger | null
+}
+
+export async function runStreamLoop(deps: StreamLoopDeps): Promise<void> {
+  const {
+    chainClient, redis, blockProcessor, backpressureGate, forkDetector,
+    supervisor, syncKey, eventsStream, irreversibleOnly, eventToFields, isStopped,
+  } = deps
+
+  const runOneConnection = async (resetBackoff: () => void): Promise<void> => {
+    if (isStopped()) return
+
+    // connect() идемпотентен при OPEN ws (первый заход — соединение уже из
+    // start-connect выше); после разрыва создаёт свежий WebSocket. handshake
+    // внутри отдаёт кешированный chainId, повторный get_status не шлётся.
+    await chainClient.connect()
+
+    const resumeNum = await redis.hget(syncKey, 'block_num')
+    const resumeId = await redis.hget(syncKey, 'block_id')
+    const havePositions =
+      resumeNum && resumeId ? [{ blockNum: Number(resumeNum), blockId: resumeId }] : []
+    const streamOpts = {
+      startBlock: havePositions[0]?.blockNum ?? 0,
+      havePositions,
+    }
+
+    try {
+      for await (const block of chainClient.streamBlocks(streamOpts)) {
+        if (isStopped()) return
+
+        // Backpressure: пока backlog стрима выше окна — ждём, не доливаем и не
+        // ack'аем (SHiP сам притормозит после max_messages_in_flight). Блок уже
+        // в RAM (≤ окна SHiP), обработаем его как только consumer освободит место.
+        if (backpressureGate) {
+          await backpressureGate.waitForCapacity(() => isStopped())
+          if (isStopped()) return
+        }
+
+        if (irreversibleOnly && block.thisBlock.blockNum > block.lastIrreversible.blockNum) {
+          chainClient.ack(1)
+          resetBackoff()
+          continue
+        }
+
+        const forkEvent = forkDetector.check(block.thisBlock.blockNum, block.thisBlock.blockId)
+        const events = await blockProcessor.process(block)
+        const toPublish: ParserEvent[] = forkEvent ? [forkEvent, ...events] : events
+
+        for (const event of toPublish) {
+          await redis.xadd(eventsStream, eventToFields(event))
+        }
+
+        await redis.hset(syncKey, {
+          block_num: String(block.thisBlock.blockNum),
+          block_id: block.thisBlock.blockId,
+          last_updated: new Date().toISOString(),
+        })
+
+        chainClient.ack(1)
+        resetBackoff()
+      }
+    } catch (err) {
+      // Штатная остановка закрывает WebSocket → for-await бросает
+      // ShipConnectionError. Это не сбой — выходим без backoff и без retry.
+      if (isStopped()) return
+      throw err
+    }
+  }
+
+  await supervisor.run(runOneConnection)
+}

packages/parser2/test/integration/backpressure.integration.test.ts

@@ -0,0 +1,80 @@
+/**
+ * Integration тест против реального Redis: backpressure-окно.
+ *
+ * Проверяем на живом Redis (chain/SHiP не нужны):
+ *   - backlog < highWater → waitForCapacity не паузит, возвращается сразу
+ *   - backlog >= highWater → пауза; после слива стрима до <= lowWater — возврат
+ *
+ * Требование: Redis на REDIS_URL (в CI всегда есть). Без Redis — тесты skip.
+ */
+
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { IoRedisStore } from '../../src/adapters/IoRedisStore.js'
+import { BackpressureGate } from '../../src/core/BackpressureGate.js'
+import type { Logger } from '../../src/logger.js'
+
+const REDIS_URL = process.env['REDIS_URL'] ?? 'redis://localhost:6379/14'
+const STREAM = '__it_bp__:stream'
+
+const silentLog = {
+  child: () => silentLog, warn: () => {}, info: () => {}, debug: () => {}, error: () => {},
+} as unknown as Logger
+
+const sleep = (ms: number): Promise<void> => new Promise(r => setTimeout(r, ms))
+
+let store: IoRedisStore
+let redisUp = false
+
+beforeAll(async () => {
+  store = new IoRedisStore({ url: REDIS_URL })
+  try {
+    await store.connect()
+    await store.client.del(STREAM)
+    redisUp = true
+  } catch {
+    redisUp = false
+  }
+})
+
+afterAll(async () => {
+  if (redisUp) {
+    await store.client.del(STREAM)
+    await store.quit()
+  }
+})
+
+describe('backpressure — окно на реальном Redis', () => {
+  it('не паузит когда backlog < highWater', async () => {
+    if (!redisUp) return
+    await store.client.del(STREAM)
+    await store.xadd(STREAM, { n: '1' })
+    const gate = new BackpressureGate({ redis: store, stream: STREAM, highWater: 5, lowWater: 2, pollMs: 10, logger: silentLog })
+    await gate.waitForCapacity()
+    expect(gate.isPaused).toBe(false)
+  })
+
+  it('паузит при backlog >= highWater и возобновляет после слива до <= lowWater', async () => {
+    if (!redisUp) return
+    await store.client.del(STREAM)
+    const ids: string[] = []
+    for (let i = 0; i < 6; i++) ids.push(await store.xadd(STREAM, { n: String(i) }))
+    expect(await store.xlen(STREAM)).toBe(6)
+
+    const gate = new BackpressureGate({ redis: store, stream: STREAM, highWater: 5, lowWater: 2, pollMs: 10, logger: silentLog })
+    let resolved = false
+    const p = gate.waitForCapacity().then(() => { resolved = true })
+
+    // даём gate войти в паузу и подождать
+    await sleep(40)
+    expect(gate.isPaused).toBe(true)
+    expect(resolved).toBe(false)
+
+    // сливаем backlog: удаляем 4 записи → XLEN=2 <= lowWater
+    await store.client.xdel(STREAM, ids[0]!, ids[1]!, ids[2]!, ids[3]!)
+    expect(await store.xlen(STREAM)).toBe(2)
+
+    await p // должно разблокироваться
+    expect(resolved).toBe(true)
+    expect(gate.isPaused).toBe(false)
+  })
+})

packages/parser2/test/unit/backpressureGate.test.ts

@@ -0,0 +1,126 @@
+/**
+ * Тесты BackpressureGate — окна поглощения между писателем и consumer'ом.
+ *
+ * Проверяем:
+ *   - валидацию highWater/lowWater в конструкторе
+ *   - мгновенный возврат когда backlog < highWater (без паузы)
+ *   - паузу при backlog >= highWater и возобновление при сливе до <= lowWater
+ *   - isPaused отражает состояние
+ *   - shouldStop() прерывает паузу (graceful shutdown) даже если backlog высок
+ */
+
+import { describe, it, expect, vi } from 'vitest'
+import { BackpressureGate } from '../../src/core/BackpressureGate.js'
+import type { RedisStore } from '../../src/ports/RedisStore.js'
+import type { Logger } from '../../src/logger.js'
+
+// Молчаливый логгер — не засоряет вывод тестов.
+const silentLog = {
+  child: () => silentLog,
+  warn: () => {},
+  info: () => {},
+  debug: () => {},
+  error: () => {},
+} as unknown as Logger
+
+function makeRedis(xlen: ReturnType<typeof vi.fn>): RedisStore {
+  return { xlen } as unknown as RedisStore
+}
+
+function makeGate(xlen: ReturnType<typeof vi.fn>, over: Partial<{ highWater: number; lowWater: number; pollMs: number }> = {}): BackpressureGate {
+  return new BackpressureGate({
+    redis: makeRedis(xlen),
+    stream: 's',
+    highWater: over.highWater ?? 100,
+    lowWater: over.lowWater ?? 50,
+    pollMs: over.pollMs ?? 1,
+    logger: silentLog,
+  })
+}
+
+describe('BackpressureGate — конструктор', () => {
+  const redis = makeRedis(vi.fn())
+
+  it('бросает если highWater <= 0', () => {
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 0, lowWater: 0 })).toThrow(/highWater/)
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: -5, lowWater: 0 })).toThrow(/highWater/)
+  })
+
+  it('бросает если lowWater >= highWater', () => {
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 100, lowWater: 100 })).toThrow(/lowWater/)
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 100, lowWater: 150 })).toThrow(/lowWater/)
+  })
+
+  it('бросает если lowWater < 0', () => {
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 100, lowWater: -1 })).toThrow(/lowWater/)
+  })
+
+  it('принимает валидные границы', () => {
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 100, lowWater: 50 })).not.toThrow()
+    expect(() => new BackpressureGate({ redis, stream: 's', highWater: 100, lowWater: 0 })).not.toThrow()
+  })
+})
+
+describe('BackpressureGate — waitForCapacity', () => {
+  it('возвращается сразу когда backlog < highWater (одна проверка XLEN, без паузы)', async () => {
+    const xlen = vi.fn().mockResolvedValue(50)
+    const gate = makeGate(xlen)
+    await gate.waitForCapacity()
+    expect(xlen).toHaveBeenCalledTimes(1)
+    expect(gate.isPaused).toBe(false)
+  })
+
+  it('возвращается сразу при backlog = highWater - 1', async () => {
+    const xlen = vi.fn().mockResolvedValue(99)
+    const gate = makeGate(xlen)
+    await gate.waitForCapacity()
+    expect(xlen).toHaveBeenCalledTimes(1)
+  })
+
+  it('ставит паузу при backlog >= highWater и возобновляет при сливе до <= lowWater', async () => {
+    // 150 (пауза) → опрос: 150, 70 (>50), 30 (<=50 → break)
+    const xlen = vi.fn()
+      .mockResolvedValueOnce(150)
+      .mockResolvedValueOnce(150)
+      .mockResolvedValueOnce(70)
+      .mockResolvedValueOnce(30)
+    const gate = makeGate(xlen)
+    await gate.waitForCapacity()
+    expect(xlen).toHaveBeenCalledTimes(4)
+    expect(gate.isPaused).toBe(false)
+  })
+
+  it('пауза срабатывает ровно на backlog == highWater (граница >=)', async () => {
+    const xlen = vi.fn().mockResolvedValueOnce(100).mockResolvedValueOnce(40)
+    const gate = makeGate(xlen)
+    await gate.waitForCapacity()
+    expect(xlen).toHaveBeenCalledTimes(2) // 100 → пауза, 40 → выход
+  })
+
+  it('isPaused = true во время паузы', async () => {
+    const snapshots: boolean[] = []
+    let gate: BackpressureGate
+    const xlen = vi.fn().mockImplementation(() => {
+      // снимок состояния паузы в момент опроса XLEN
+      snapshots.push(gate.isPaused)
+      return Promise.resolve(snapshots.length >= 3 ? 10 : 200)
+    })
+    gate = makeGate(xlen)
+    await gate.waitForCapacity()
+    // 1-й вызов — до установки паузы (false); внутри цикла — true
+    expect(snapshots[0]).toBe(false)
+    expect(snapshots.some(s => s === true)).toBe(true)
+    expect(gate.isPaused).toBe(false)
+  })
+
+  it('shouldStop() прерывает паузу даже при высоком backlog (graceful shutdown)', async () => {
+    const xlen = vi.fn().mockResolvedValue(9999) // никогда не сольётся
+    const gate = makeGate(xlen)
+    let stop = false
+    const p = gate.waitForCapacity(() => stop)
+    // через тик включаем стоп
+    setTimeout(() => { stop = true }, 5)
+    await p // не должно зависнуть
+    expect(gate.isPaused).toBe(false)
+  })
+})

packages/parser2/test/unit/reconnectSupervisor.test.ts

@@ -71,3 +71,69 @@ describe('ReconnectSupervisor — exhaustion', () => {
     delays.forEach(d => expect(d).toBe(0))
   })
 })
+
+describe('ReconnectSupervisor — resetBackoff (прогресс сбрасывает счётчик)', () => {
+  it('передаёт resetBackoff в fn', async () => {
+    const sup = new ReconnectSupervisor({ backoffSeconds: [0] })
+    let gotCallback = false
+    await sup.run(async (reset) => {
+      gotCallback = typeof reset === 'function'
+      return 'ok'
+    })
+    expect(gotCallback).toBe(true)
+  })
+
+  it('не вызывает onGiveUp пока fn делает прогресс между разрывами', async () => {
+    // Сценарий: соединение каждый раз отдаёт «блок» (прогресс), затем рвётся.
+    // Без сброса счётчик дошёл бы до maxAttempts=3 и завершил процесс. С resetBackoff
+    // на каждом прогрессе счётчик возвращается в 0 — даём 10 циклов разрыва, всё живёт.
+    let gaveUp = false
+    let drops = 0
+    const sup = new ReconnectSupervisor({
+      maxAttempts: 3,
+      backoffSeconds: [0],
+      onGiveUp: () => { gaveUp = true; throw new Error('gave up') },
+    })
+    await sup.run(async (reset) => {
+      reset() // обработали блок — прогресс есть
+      drops++
+      if (drops < 10) throw new Error('node dropped') // разрыв
+      return 'done'
+    })
+    expect(gaveUp).toBe(false)
+    expect(drops).toBe(10)
+  })
+
+  it('БЕЗ прогресса исчерпывает попытки и вызывает onGiveUp', async () => {
+    // Контроль к предыдущему: мёртвая нода (resetBackoff не зовётся) → exit.
+    let gaveUp = false
+    const sup = new ReconnectSupervisor({
+      maxAttempts: 3,
+      backoffSeconds: [0],
+      onGiveUp: () => { gaveUp = true; throw new Error('gave up') },
+    })
+    await expect(
+      sup.run(async () => { throw new Error('dead node') }),
+    ).rejects.toThrow()
+    expect(gaveUp).toBe(true)
+  })
+
+  it('resetBackoff возвращает нумерацию onAttempt в начало', async () => {
+    const attempts: number[] = []
+    const sup = new ReconnectSupervisor({
+      maxAttempts: 10,
+      backoffSeconds: [0],
+      onAttempt: (attempt) => attempts.push(attempt),
+    })
+    let calls = 0
+    await sup.run(async (reset) => {
+      calls++
+      // calls: 1 throw → attempt 1; 2 throw → attempt 2; 3 reset+throw → attempt 1 снова; 4 ok
+      if (calls === 3) reset()
+      if (calls < 4) throw new Error('fail')
+      return 'ok'
+    })
+    // attempt-числа: [1, 2, 1] — третий разрыв после reset снова стартует с 1
+    expect(attempts).toEqual([1, 2, 1])
+  })
+})

packages/parser2/test/unit/streamLoop.test.ts

@@ -0,0 +1,160 @@
+/**
+ * Юнит-тесты цикла чтения с авто-переподключением (runStreamLoop) на фейках —
+ * без реального Redis и SHiP-ноды.
+ *
+ * Покрываем то, ради чего цикл вынесен из Parser:
+ *   1. разрыв ноды → backoff → переподключение, чтение продолжается;
+ *   2. после разрыва позиция возобновления берётся из Redis (последний блок) —
+ *      без потерь и дублей;
+ *   3. штатная остановка не вызывает reconnect;
+ *   4. полностью мёртвая нода исчерпывает попытки и завершается (onGiveUp),
+ *      а НЕ долбит без пауз.
+ */
+
+import { describe, it, expect } from 'vitest'
+import type { GetBlocksOptions, ShipBlock } from '@coopenomics/coopos-ship-reader'
+import { runStreamLoop } from '../../src/core/streamLoop.js'
+import type { StreamLoopChainClient, StreamLoopRedis } from '../../src/core/streamLoop.js'
+import { ReconnectSupervisor } from '../../src/core/ReconnectSupervisor.js'
+
+function makeBlock(n: number): ShipBlock {
+  return {
+    thisBlock: { blockNum: n, blockId: `id${n}` },
+    lastIrreversible: { blockNum: n, blockId: `id${n}` },
+    head: { blockNum: n, blockId: `id${n}` },
+    prevBlock: null,
+    blockTime: '2024-06-01T00:00:00.000',
+    traces: [],
+    deltas: [],
+  } as unknown as ShipBlock
+}
+
+function makeRedis(): StreamLoopRedis & { hash: Record<string, Record<string, string>> } {
+  const hash: Record<string, Record<string, string>> = {}
+  return {
+    hash,
+    async hget(key, field) { return hash[key]?.[field] ?? null },
+    async hset(key, fields) { hash[key] = { ...(hash[key] ?? {}), ...fields }; return 'OK' },
+    async xadd() { return '1-0' },
+  }
+}
+
+/** Скриптованный SHiP-клиент: массив сессий, каждая — список блоков + рвётся/нет. */
+function makeChainClient(sessions: Array<{ blocks: number[]; drop: boolean }>) {
+  const captured: GetBlocksOptions[] = []
+  let session = 0
+  let connectCalls = 0
+  const client: StreamLoopChainClient = {
+    async connect() { connectCalls++ },
+    ack() {},
+    async *streamBlocks(opts: GetBlocksOptions) {
+      captured.push(opts)
+      const s = sessions[session++]
+      if (!s) return
+      for (const n of s.blocks) yield makeBlock(n)
+      if (s.drop) throw new Error('node dropped')
+    },
+  }
+  return {
+    client,
+    captured,
+    get connectCalls() { return connectCalls },
+  }
+}
+
+const baseDeps = (over: Partial<Parameters<typeof runStreamLoop>[0]>): Parameters<typeof runStreamLoop>[0] => ({
+  chainClient: over.chainClient!,
+  redis: over.redis!,
+  blockProcessor: over.blockProcessor ?? { process: async () => [] },
+  backpressureGate: null,
+  forkDetector: { check: () => null },
+  supervisor: over.supervisor ?? new ReconnectSupervisor({ maxAttempts: 5, backoffSeconds: [0] }),
+  syncKey: 'sync',
+  eventsStream: 'events',
+  irreversibleOnly: false,
+  eventToFields: () => ({ data: '{}' }),
+  isStopped: over.isStopped ?? (() => false),
+  log: null,
+  ...over,
+})
+
+describe('runStreamLoop — переподключение после разрыва', () => {
+  it('после разрыва переподключается и продолжает с последнего записанного блока', async () => {
+    const redis = makeRedis()
+    const cc = makeChainClient([
+      { blocks: [1, 2, 3], drop: true },  // упала после блока 3
+      { blocks: [4, 5], drop: false },    // реконнект → дочитали и штатно завершились
+    ])
+
+    await runStreamLoop(baseDeps({ chainClient: cc.client, redis }))
+
+    // Подключались дважды: первичное + 1 реконнект
+    expect(cc.connectCalls).toBe(2)
+    expect(cc.captured).toHaveLength(2)
+    // Первая сессия — с нуля (позиции в Redis ещё нет)
+    expect(cc.captured[0]?.startBlock).toBe(0)
+    // Вторая — возобновление ровно с блока 3 (последнего записанного), без потерь/дублей
+    expect(cc.captured[1]?.startBlock).toBe(3)
+    expect(cc.captured[1]?.havePositions).toEqual([{ blockNum: 3, blockId: 'id3' }])
+    // Финальная позиция — блок 5
+    expect(redis.hash['sync']?.['block_num']).toBe('5')
+  })
+
+  it('штатная остановка не вызывает reconnect', async () => {
+    const redis = makeRedis()
+    const cc = makeChainClient([{ blocks: [1, 2, 3, 4], drop: false }])
+    let processed = 0
+    const deps = baseDeps({
+      chainClient: cc.client,
+      redis,
+      blockProcessor: { process: async () => { processed++; return [] } },
+      isStopped: () => processed >= 2, // останавливаемся после 2 блоков
+    })
+
+    await runStreamLoop(deps)
+
+    expect(cc.connectCalls).toBe(1)      // без переподключений
+    expect(redis.hash['sync']?.['block_num']).toBe('2')
+  })
+
+  it('мёртвая нода исчерпывает попытки и вызывает onGiveUp (не долбит без пауз)', async () => {
+    const redis = makeRedis()
+    const cc = makeChainClient([
+      { blocks: [], drop: true },
+      { blocks: [], drop: true },
+      { blocks: [], drop: true },
+      { blocks: [], drop: true },
+    ])
+    let gaveUp = false
+    const supervisor = new ReconnectSupervisor({
+      maxAttempts: 3,
+      backoffSeconds: [0],
+      onGiveUp: () => { gaveUp = true; throw new Error('gave up') },
+    })
+
+    await expect(
+      runStreamLoop(baseDeps({ chainClient: cc.client, redis, supervisor })),
+    ).rejects.toThrow()
+    expect(gaveUp).toBe(true)
+  })
+
+  it('долгая стабильная сессия после редкого разрыва не копит попытки до выхода', async () => {
+    // 6 разрывов подряд, но каждая сессия отдаёт блок (прогресс) → resetBackoff
+    // держит счётчик у нуля. maxAttempts=3 НЕ срабатывает, т.к. между разрывами прогресс.
+    const redis = makeRedis()
+    const sessions = Array.from({ length: 6 }, (_, i) => ({ blocks: [i + 1], drop: i < 5 }))
+    const cc = makeChainClient(sessions)
+    let gaveUp = false
+    const supervisor = new ReconnectSupervisor({
+      maxAttempts: 3,
+      backoffSeconds: [0],
+      onGiveUp: () => { gaveUp = true; throw new Error('gave up') },
+    })
+
+    await runStreamLoop(baseDeps({ chainClient: cc.client, redis, supervisor }))
+
+    expect(gaveUp).toBe(false)
+    expect(cc.connectCalls).toBe(6)
+    expect(redis.hash['sync']?.['block_num']).toBe('6')
+  })
+})