Merge pull request 'fix(xtrim): не триммить un-acked pending + null-гард parseStreamEntries (баг #3)' (#9) from dev into main

Баг #3: consumer падал навсегда с TypeError "Cannot read properties of null
(reading 'length')" и переставал читать стрим. Два корня:

1. XtrimSupervisor.trim() брал MINID = lastDeliveredId. Но un-acked pending
   записи старше lastDeliveredId (доставлены, не подтверждены) → XTRIM сносил
   собственные pending группы. Теперь нижняя граница trim — самый старый
   un-acked ID из XPENDING (новый метод RedisStore.xpendingMinId), а сравнение
   stream-ID числовое по <ms>-<seq>, не лексикографическое ('1000' < '999').

2. parseStreamEntries падал на null rawFields: при перечитывании PEL
   (XREADGROUP ... 0) Redis отдаёт [id, null] для обрезанных ID. Добавлен
   гард `if (!rawFields) continue` — обрезанные записи пропускаются.

Триггер: consumer отстаёт от писателя (бутстрап с genesis) → растёт pending →
XtrimSupervisor сносит pending → PEL отдаёт null → краш в бесконечном цикле.

Воспроизведено и проверено на реальном Redis (новый integration-тест
xtrim.integration.test.ts, гоняется в CI): trim сохраняет все un-acked pending,
PEL-перечитка не падает на null. Unit: 211 зелёных.

parser2 1.1.2→1.1.3. ship-reader без изменений.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/parser2/package.json

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

packages/parser2/src/adapters/IoRedisStore.ts

@@ -37,7 +37,9 @@ interface IRedisClient {
     count: string, countVal: number,
     block: string, blockMs: number,
     streams: string, stream: string, id: string,
-  ): Promise<Array<[string, Array<[string, string[]]>]> | null>
+  ): Promise<Array<[string, Array<[string, string[] | null]>]> | null>
+  // XPENDING key group (summary): [count, min-id, max-id, [[consumer, count],...]|null]
+  xpending(key: string, group: string): Promise<[number, string | null, string | null, Array<[string, string]> | null] | null>
   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>
@@ -94,9 +96,14 @@ return 0
  * Redis возвращает: [[id, [field1, val1, field2, val2, ...]], ...]
  * Мы конвертируем в: [{id, fields: {field1: val1, ...}}, ...]
  */
-function parseStreamEntries(raw: Array<[string, string[]]>): StreamMessage[] {
+function parseStreamEntries(raw: Array<[string, string[] | null]>): StreamMessage[] {
   const messages: StreamMessage[] = []
   for (const [msgId, rawFields] of raw) {
+    // Обрезанные/удалённые записи: при чтении PEL (XREADGROUP ... 0) Redis
+    // отдаёт [id, null] для ID, которых уже нет в стриме (XTRIM/XDEL их снёс).
+    // rawFields === null → .length бросает TypeError и убивает consumer навсегда.
+    // Пропускаем — запись недоступна, восстановить нечего.
+    if (!rawFields) continue
     const fields: Record<string, string> = {}
     // rawFields — плоский массив [key, val, key, val, ...], шагаем по 2
     for (let i = 0; i + 1 < rawFields.length; i += 2) {
@@ -246,6 +253,19 @@ export class IoRedisStore implements RedisStore {
     await this.client.xack(stream, group, id)
   }
 
+  /**
+   * XPENDING stream group (summary-форма) → самый старый un-acked ID в PEL группы.
+   * null если pending нет. Нужен для безопасного XTRIM: триммить можно только
+   * НИЖЕ этого ID, иначе снесём собственные недоставленные/неподтверждённые записи.
+   */
+  async xpendingMinId(stream: string, group: string): Promise<string | null> {
+    const res = await this.client.xpending(stream, group)
+    if (!Array.isArray(res)) return null
+    const [count, minId] = res
+    if (!count || !minId) return null
+    return String(minId)
+  }
+
   /** ZADD key score member. */
   async zadd(key: string, score: number, member: string): Promise<void> {
     await this.client.zadd(key, score, member)

packages/parser2/src/core/BlockProcessor.ts

@@ -97,8 +97,13 @@ export class BlockProcessor {
 
     const blockNum = block.thisBlock.blockNum
     const blockId = block.thisBlock.blockId
-    // blockTime берём из первой трассировки; если трассировок нет — текущее время
-    const blockTime = block.traces[0]?.blockTime ?? new Date().toISOString()
+    // block_time — on-chain время блока (signed_block.timestamp), общее для всех событий
+    // блока, включая trace-less блоки (genesis). НЕ wall-clock: подставлять new Date()
+    // нельзя — это ломает point-in-time запросы. Пустое = block не запрошен у SHiP.
+    const blockTime = block.blockTime
+    if (!blockTime) {
+      this.log.warn({ block_num: blockNum }, 'block_time missing from ship block (fetch_block disabled?)')
+    }
 
     // ── Фаза 1: Action traces ─────────────────────────────────────────────────
     for (const trace of block.traces) {

packages/parser2/src/core/XtrimSupervisor.ts

@@ -7,9 +7,15 @@
  * Стратегия MINID:
  *   Вместо хранения фиксированного числа записей (MAXLEN), мы сохраняем все
  *   записи, которые ещё не подтверждены (pending) хотя бы одной consumer group.
- *   minId = min(lastDeliveredId всех групп с pending > 0).
+ *   minId = min(самый старый un-acked pending ID всех групп с pending > 0).
  *   XTRIM stream MINID minId удаляет всё с ID < minId.
  *
+ * ВАЖНО: триммить по lastDeliveredId НЕЛЬЗЯ — un-acked pending записи старше
+ * lastDeliveredId (доставлены, но не подтверждены: их ID < last-delivered).
+ * XTRIM MINID lastDeliveredId снёс бы собственные pending группы → consumer при
+ * перечитывании PEL получает [id, null] и падает. Поэтому нижняя граница trim —
+ * самый старый un-acked ID из XPENDING, а не lastDeliveredId.
+ *
  * Это гарантирует, что ни один consumer не потеряет сообщения при trim:
  * группа с отставанием «тормозит» trim, пока не догонит.
  *
@@ -18,6 +24,19 @@
 
 import type { RedisStore } from '../ports/RedisStore.js'
 
+/**
+ * Числовое сравнение Redis Stream ID формата `<ms>-<seq>`.
+ * Строковое сравнение (a < b) неверно: '1000-0' < '999-0' лексикографически
+ * true, хотя по времени 1000 > 999. Возвращает <0, 0, >0.
+ */
+function compareStreamIds(a: string, b: string): number {
+  const [aMs, aSeq = '0'] = a.split('-')
+  const [bMs, bSeq = '0'] = b.split('-')
+  const ms = Number(aMs) - Number(bMs)
+  if (ms !== 0) return ms
+  return Number(aSeq) - Number(bSeq)
+}
+
 export interface XtrimSupervisorOpts {
   redis: RedisStore
   /** Имя стрима для очистки (обычно ce:parser2:<chainId>:events). */
@@ -60,8 +79,8 @@ export class XtrimSupervisor {
    * Один цикл очистки:
    * 1. Получаем список consumer groups через XINFO GROUPS.
    * 2. Фильтруем группы у которых есть pending сообщения (pending > 0).
-   * 3. Находим минимальный lastDeliveredId среди таких групп.
-   * 4. XTRIM stream MINID minId — удаляем всё старее этого ID.
+   * 3. Для каждой берём самый старый un-acked pending ID (XPENDING).
+   * 4. minId = числовой минимум этих ID; XTRIM stream MINID minId.
    *
    * Если pending-групп нет — trim не делается (всё подтверждено).
    * Если стрим не существует или XInfo бросает — тихо игнорируем (best-effort).
@@ -75,12 +94,19 @@ export class XtrimSupervisor {
       const pendingGroups = groups.filter(g => g.pending > 0)
       if (pendingGroups.length === 0) return
 
-      // Наименьший lastDeliveredId = самый отстающий consumer
-      const minId = pendingGroups
-        .map(g => g.lastDeliveredId)
-        .reduce((a, b) => (a < b ? a : b))
+      // Самый старый un-acked pending ID по каждой группе. НЕ lastDeliveredId:
+      // он выше pending, trim по нему снёс бы собственные неподтверждённые записи.
+      const oldestPending: string[] = []
+      for (const g of pendingGroups) {
+        const id = await this.redis.xpendingMinId(this.stream, g.name)
+        if (id) oldestPending.push(id)
+      }
+      if (oldestPending.length === 0) return
 
-      if (minId) await this.redis.xtrim(this.stream, minId)
+      // Нижняя граница trim = самый старый un-acked ID среди всех групп
+      const minId = oldestPending.reduce((a, b) => (compareStreamIds(a, b) <= 0 ? a : b))
+
+      await this.redis.xtrim(this.stream, minId)
     } catch {
       // XTRIM — best-effort: ошибки не должны влиять на основной поток обработки
     }

packages/parser2/src/ports/RedisStore.ts

@@ -79,6 +79,13 @@ export interface RedisStore {
   /** XACK stream group id — подтверждает обработку, убирает из PEL. */
   xack(stream: string, group: string, id: string): Promise<void>
 
+  /**
+   * XPENDING stream group → ID самого старого un-acked сообщения в PEL группы
+   * (или null если pending нет). Используется XtrimSupervisor: trim допустим
+   * только НИЖЕ этого ID, иначе un-acked записи будут потеряны.
+   */
+  xpendingMinId(stream: string, group: string): Promise<string | null>
+
   // ── Sorted Set ────────────────────────────────────────────────────────────
 
   /** ZADD key score member. */

packages/parser2/test/integration/xtrim.integration.test.ts

@@ -0,0 +1,103 @@
+/**
+ * Integration тест против реального Redis: регресс-гард на баг #3.
+ *
+ * Баг: XtrimSupervisor триммил по lastDeliveredId и сносил собственные un-acked
+ * pending записи; при перечитывании PEL Redis отдавал [id, null], а
+ * parseStreamEntries падал на null.length → consumer умирал навсегда.
+ *
+ * Здесь воспроизводим точный путь на живом Redis (chain/SHiP не нужны):
+ *   A. trim по lastDeliveredId зануляет pending → xreadGroup(PEL) НЕ падает на null.
+ *   B. XtrimSupervisor.trim() (новая логика) сохраняет все un-acked pending.
+ *
+ * Требование: Redis на REDIS_URL (в CI всегда есть). Без Redis — тесты skip.
+ */
+
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { IoRedisStore } from '../../src/adapters/IoRedisStore.js'
+import { XtrimSupervisor } from '../../src/core/XtrimSupervisor.js'
+
+const REDIS_URL = process.env['REDIS_URL'] ?? 'redis://localhost:6379/14'
+const STREAM = '__it_bug3__:stream'
+const GROUP = 'g'
+
+let store: IoRedisStore
+let redisUp = false
+
+// XtrimSupervisor.trim приватный — дёргаем один цикл напрямую через каст.
+function runTrimOnce(sup: XtrimSupervisor): Promise<void> {
+  return (sup as unknown as { trim(): Promise<void> }).trim()
+}
+
+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('bug #3 — XTRIM/PEL на реальном Redis', () => {
+  it('xreadGroup(PEL) не падает на обрезанных [id,null] записях', async () => {
+    if (!redisUp) return
+    await store.client.del(STREAM)
+    await store.xgroupCreate(STREAM, GROUP, '0')
+    for (let i = 0; i < 10; i++) await store.xadd(STREAM, { kind: 'action', n: String(i) })
+
+    // читаем 5 → они pending (un-acked); lastDeliveredId = id 5-й записи
+    const read = await store.xreadGroup(STREAM, GROUP, 'c1', 5, 0, '>')
+    expect(read).toHaveLength(5)
+    const lastDelivered = read[4]!.id
+
+    // ИМИТАЦИЯ СТАРОГО БАГА: trim по lastDeliveredId сносит первые 4 pending
+    await store.client.xtrim(STREAM, 'MINID', lastDelivered)
+
+    // consumer перечитывает PEL — Redis отдаёт [id,null] для снесённых.
+    // Не должно бросать (старый parseStreamEntries падал на null.length).
+    const pel = await store.xreadGroup(STREAM, GROUP, 'c1', 100, 0, '0')
+    // обрезанные пропущены, выжившие — с полями
+    expect(pel.every(m => Object.keys(m.fields).length > 0)).toBe(true)
+    expect(pel.some(m => m.id === lastDelivered)).toBe(true)
+  })
+
+  it('XtrimSupervisor.trim() сохраняет все un-acked pending записи', async () => {
+    if (!redisUp) return
+    await store.client.del(STREAM)
+    await store.xgroupCreate(STREAM, GROUP, '0')
+    for (let i = 0; i < 10; i++) await store.xadd(STREAM, { kind: 'action', n: String(i) })
+
+    const read = await store.xreadGroup(STREAM, GROUP, 'c1', 5, 0, '>')
+    expect(read).toHaveLength(5)
+    const oldestPending = read[0]!.id
+
+    const sup = new XtrimSupervisor({ redis: store, stream: STREAM, intervalMs: 999_999 })
+    await runTrimOnce(sup)
+
+    // consumer перечитывает PEL — все 5 pending целы, без null, без краша
+    const pel = await store.xreadGroup(STREAM, GROUP, 'c1', 100, 0, '0')
+    expect(pel).toHaveLength(5)
+    expect(pel.every(m => Object.keys(m.fields).length > 0)).toBe(true)
+    // нижняя граница trim = oldestPending → первый pending на месте
+    expect(pel[0]!.id).toBe(oldestPending)
+  })
+
+  it('xpendingMinId возвращает самый старый un-acked ID, null без pending', async () => {
+    if (!redisUp) return
+    await store.client.del(STREAM)
+    await store.xgroupCreate(STREAM, GROUP, '0')
+    expect(await store.xpendingMinId(STREAM, GROUP)).toBeNull()
+
+    for (let i = 0; i < 3; i++) await store.xadd(STREAM, { kind: 'action', n: String(i) })
+    const read = await store.xreadGroup(STREAM, GROUP, 'c1', 3, 0, '>')
+    expect(await store.xpendingMinId(STREAM, GROUP)).toBe(read[0]!.id)
+  })
+})

packages/parser2/test/unit/blockProcessor.test.ts

@@ -51,6 +51,7 @@ function makeBlock(blockNum = 1, numTraces = 0, numDeltas = 0): ShipBlock {
     head: blockPosition,
     lastIrreversible: blockPosition,
     prevBlock: null,
+    blockTime: '2024-06-01T12:00:00.000',
     traces: Array.from({ length: numTraces }, (_, i) => ({
       account: 'eosio.token',
       name: 'transfer',
@@ -175,6 +176,7 @@ describe('BlockProcessor — ABI updates (Story 4.3)', () => {
       head: { blockNum: 500, blockId: 'c'.repeat(64) },
       lastIrreversible: { blockNum: 500, blockId: 'c'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [{
         account: 'eosio',
         name: 'setabi',
@@ -214,6 +216,7 @@ describe('BlockProcessor — ABI updates (Story 4.3)', () => {
       head: { blockNum: 501, blockId: 'c'.repeat(64) },
       lastIrreversible: { blockNum: 501, blockId: 'c'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [{
         account: 'eosio',
         name: 'setabi',
@@ -253,6 +256,7 @@ describe('BlockProcessor — ABI updates (Story 4.3)', () => {
       head: { blockNum: 600, blockId: 'e'.repeat(64) },
       lastIrreversible: { blockNum: 600, blockId: 'e'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [],
       deltas: [{
         name: 'account' as never,
@@ -283,6 +287,7 @@ describe('BlockProcessor — native deltas (Epic 6)', () => {
       head: { blockNum: 10, blockId: 'a'.repeat(64) },
       lastIrreversible: { blockNum: 10, blockId: 'a'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [],
       deltas: nativeTableNames.map(name => ({
         name: name as never,
@@ -338,6 +343,7 @@ describe('BlockProcessor — native deltas (Epic 6)', () => {
       head: { blockNum: 20, blockId: 'b'.repeat(64) },
       lastIrreversible: { blockNum: 20, blockId: 'b'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [
         { account: 'eosio.token', name: 'transfer', authorization: [], actRaw: new Uint8Array([1]), actionOrdinal: 1, creatorActionOrdinal: 0, globalSequence: BigInt(1), receipt: null, contextFree: false, elapsed: 0, console: '', accountRamDeltas: [], blockNum: 20, blockId: 'b'.repeat(64), blockTime: '2024-01-01T00:00:00.000', transactionId: 'c'.repeat(64) },
         { account: 'eosio.token', name: 'transfer', authorization: [], actRaw: new Uint8Array([1]), actionOrdinal: 2, creatorActionOrdinal: 0, globalSequence: BigInt(2), receipt: null, contextFree: false, elapsed: 0, console: '', accountRamDeltas: [], blockNum: 20, blockId: 'b'.repeat(64), blockTime: '2024-01-01T00:00:00.000', transactionId: 'c'.repeat(64) },
@@ -370,6 +376,7 @@ describe('BlockProcessor — native deltas (Epic 6)', () => {
       head: { blockNum: 5, blockId: 'd'.repeat(64) },
       lastIrreversible: { blockNum: 5, blockId: 'd'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [{ account: 'eosio', name: 'newaccount', authorization: [], actRaw: new Uint8Array([1]), actionOrdinal: 1, creatorActionOrdinal: 0, globalSequence: BigInt(10), receipt: null, contextFree: false, elapsed: 0, console: '', accountRamDeltas: [], blockNum: 5, blockId: 'd'.repeat(64), blockTime: '2024-01-01T00:00:00.000', transactionId: 'e'.repeat(64) }],
       deltas: [
         { name: 'contract_row', present: true, rowRaw: new Uint8Array([1]), code: 'eosio', scope: 'eosio', table: 'global', primaryKey: '0' },
@@ -390,6 +397,7 @@ describe('BlockProcessor — native deltas (Epic 6)', () => {
       head: { blockNum: 700, blockId: 'f'.repeat(64) },
       lastIrreversible: { blockNum: 700, blockId: 'f'.repeat(64) },
       prevBlock: null,
+      blockTime: '2024-06-01T12:00:00.000',
       traces: [],
       deltas: [{ name: 'account' as never, present: true, rowRaw: new Uint8Array([1, 2, 3]) }],
     }

packages/parser2/test/unit/ioRedisStore.test.ts

@@ -5,6 +5,8 @@ vi.mock('ioredis', () => {
   const mockRedis = {
     xadd: vi.fn().mockResolvedValue('1700000000000-0'),
     xtrim: vi.fn().mockResolvedValue(5),
+    xreadgroup: vi.fn().mockResolvedValue(null),
+    xpending: vi.fn().mockResolvedValue([0, null, null, null]),
     zadd: vi.fn().mockResolvedValue(1),
     zrangebyscore: vi.fn().mockResolvedValue(['{"version":"eosio::abi/1.0"}']),
     zrevrangebyscore: vi.fn().mockResolvedValue(['{"version":"eosio::abi/1.0"}']),
@@ -37,6 +39,34 @@ describe('IoRedisStore', () => {
     expect(store.client.xtrim).toHaveBeenCalledWith('mystream', 'MINID', '1234567890000-0')
   })
 
+  it('xreadGroup skips trimmed entries with null fields (no crash — баг #3)', async () => {
+    // Redis отдаёт [id, null] для обрезанных/удалённых ID при чтении PEL.
+    // Раньше rawFields.length на null убивал consumer навсегда.
+    vi.mocked(store.client.xreadgroup).mockResolvedValueOnce([
+      ['mystream', [
+        ['100-0', ['kind', 'action']],
+        ['101-0', null],            // обрезанная запись (XTRIM/XDEL)
+        ['102-0', ['kind', 'delta']],
+      ]],
+    ])
+    const msgs = await store.xreadGroup('mystream', 'g', 'c', 10, 0, '0')
+    expect(msgs).toHaveLength(2)
+    expect(msgs.map(m => m.id)).toEqual(['100-0', '102-0'])
+    expect(msgs[0]!.fields).toEqual({ kind: 'action' })
+  })
+
+  it('xpendingMinId returns oldest pending id from XPENDING summary', async () => {
+    vi.mocked(store.client.xpending).mockResolvedValueOnce([5, '90-0', '100-0', [['c', '5']]])
+    const id = await store.xpendingMinId('mystream', 'g')
+    expect(id).toBe('90-0')
+  })
+
+  it('xpendingMinId returns null when group has no pending', async () => {
+    vi.mocked(store.client.xpending).mockResolvedValueOnce([0, null, null, null])
+    const id = await store.xpendingMinId('mystream', 'g')
+    expect(id).toBeNull()
+  })
+
   it('zadd calls redis.zadd with score and member', async () => {
     await store.zadd('parser2:abi:eosio', 100, '{"version":"eosio::abi/1.0"}')
     expect(store.client.zadd).toHaveBeenCalledWith('parser2:abi:eosio', 100, '{"version":"eosio::abi/1.0"}')

packages/parser2/test/unit/parserClient.test.ts

@@ -69,6 +69,7 @@ class FakeRedis implements RedisStore {
   xlen = vi.fn(async (): Promise<number> => 0)
   xdel = vi.fn(async (): Promise<number> => 0)
   xack = vi.fn(async (): Promise<void> => {})
+  xpendingMinId = vi.fn(async (): Promise<string | null> => null)
 
   zadd = vi.fn(async (key: string, score: number, member: string): Promise<void> => {
     if (!this.zsets.has(key)) this.zsets.set(key, new Map())

packages/parser2/test/unit/xtrimSupervisor.test.ts

@@ -3,20 +3,28 @@
  *
  * Проверяем:
  *   - start/stop идемпотентны
- *   - trim использует MINID по отстающей группе с pending > 0
+ *   - trim использует MINID по самому старому un-acked pending ID (XPENDING),
+ *     а НЕ по lastDeliveredId (иначе сносит собственные pending — баг #3)
+ *   - числовое сравнение ID (а не лексикографическое)
  *   - группы без pending не мешают trim'у
- *   - XTRIM не вызывается если групп нет или ни у одной нет pending
- *   - ошибки в xinfoGroups не прерывают supervisor
+ *   - XTRIM не вызывается если групп нет, ни у одной нет pending, или pending-ID null
+ *   - ошибки в xinfoGroups/xtrim не прерывают supervisor
  */
 
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
 import { XtrimSupervisor } from '../../src/core/XtrimSupervisor.js'
 import type { RedisStore, XGroupInfo } from '../../src/ports/RedisStore.js'
 
-function makeRedis(groups: XGroupInfo[] = []): RedisStore {
+function makeRedis(
+  groups: XGroupInfo[] = [],
+  pendingIds: Record<string, string | null> = {},
+): RedisStore {
   return {
     xinfoGroups: vi.fn().mockResolvedValue(groups),
     xtrim: vi.fn().mockResolvedValue(0),
+    xpendingMinId: vi.fn().mockImplementation((_stream: string, group: string) =>
+      Promise.resolve(pendingIds[group] ?? null),
+    ),
     // остальные методы не используются в supervisor
   } as unknown as RedisStore
 }
@@ -25,6 +33,7 @@ function makeRedis(groups: XGroupInfo[] = []): RedisStore {
 async function flushPromises(): Promise<void> {
   await Promise.resolve()
   await Promise.resolve()
+  await Promise.resolve()
 }
 
 describe('XtrimSupervisor — lifecycle', () => {
@@ -124,36 +133,95 @@ describe('XtrimSupervisor — trim logic', () => {
     sup.stop()
   })
 
-  it('trims to minimum lastDeliveredId among groups with pending > 0', async () => {
-    // g1 pending но отстал на 100 — должен стать minId
-    // g2 pending и догнал до 500 — игнорируется для выбора min
-    // g3 нет pending — вообще не учитывается
-    const redis = makeRedis([
-      { name: 'g1', pending: 5, lastDeliveredId: '100-0', lag: 5, consumers: 1 },
-      { name: 'g2', pending: 2, lastDeliveredId: '500-0', lag: 2, consumers: 1 },
-      { name: 'g3', pending: 0, lastDeliveredId: '50-0', lag: 0, consumers: 1 },
-    ])
+  it('trims to the OLDEST un-acked pending ID (XPENDING), NOT lastDeliveredId', async () => {
+    // Регресс-гард на баг #3: pending-записи старше lastDeliveredId.
+    // g1: lastDelivered=100-0, но самый старый un-acked pending = 90-0.
+    // Триммить надо по 90-0, иначе un-acked 90..99 теряются и consumer падает.
+    const redis = makeRedis(
+      [{ name: 'g1', pending: 5, lastDeliveredId: '100-0', lag: 5, consumers: 1 }],
+      { g1: '90-0' },
+    )
     const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })
     sup.start()
     await vi.advanceTimersByTimeAsync(100)
     await flushPromises()
 
     expect(redis.xtrim).toHaveBeenCalledTimes(1)
-    expect(redis.xtrim).toHaveBeenCalledWith('s', '100-0')
+    expect(redis.xtrim).toHaveBeenCalledWith('s', '90-0')
+    // именно НЕ lastDeliveredId
+    expect(redis.xtrim).not.toHaveBeenCalledWith('s', '100-0')
     sup.stop()
   })
 
-  it('uses lexicographic comparison for stream IDs (which is also numeric-correct for same-length)', async () => {
-    const redis = makeRedis([
-      { name: 'g1', pending: 1, lastDeliveredId: '1000-0', lag: 1, consumers: 1 },
-      { name: 'g2', pending: 1, lastDeliveredId: '999-0', lag: 1, consumers: 1 },
-    ])
+  it('trims to minimum oldest-pending-id among groups with pending > 0', async () => {
+    // g1 отстал (oldest pending 90-0) → станет minId
+    // g2 догнал (oldest pending 480-0) → не минимум
+    // g3 нет pending → не учитывается (xpendingMinId не зовётся)
+    const redis = makeRedis(
+      [
+        { name: 'g1', pending: 5, lastDeliveredId: '100-0', lag: 5, consumers: 1 },
+        { name: 'g2', pending: 2, lastDeliveredId: '500-0', lag: 2, consumers: 1 },
+        { name: 'g3', pending: 0, lastDeliveredId: '50-0', lag: 0, consumers: 1 },
+      ],
+      { g1: '90-0', g2: '480-0' },
+    )
     const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })
     sup.start()
     await vi.advanceTimersByTimeAsync(100)
     await flushPromises()
-    // lexicographic: '1000-0' < '999-0' → выбран '1000-0'
-    expect(redis.xtrim).toHaveBeenCalledWith('s', '1000-0')
+
+    expect(redis.xtrim).toHaveBeenCalledTimes(1)
+    expect(redis.xtrim).toHaveBeenCalledWith('s', '90-0')
+    // g3 без pending — XPENDING не запрашивался
+    expect(redis.xpendingMinId).not.toHaveBeenCalledWith('s', 'g3')
+    sup.stop()
+  })
+
+  it('uses NUMERIC (not lexicographic) comparison for stream IDs', async () => {
+    // Лексикографически '1000-0' < '999-0' (неверно). Численно 999 < 1000.
+    const redis = makeRedis(
+      [
+        { name: 'g1', pending: 1, lastDeliveredId: '1000-0', lag: 1, consumers: 1 },
+        { name: 'g2', pending: 1, lastDeliveredId: '999-0', lag: 1, consumers: 1 },
+      ],
+      { g1: '1000-0', g2: '999-0' },
+    )
+    const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })
+    sup.start()
+    await vi.advanceTimersByTimeAsync(100)
+    await flushPromises()
+    // численный минимум = '999-0'
+    expect(redis.xtrim).toHaveBeenCalledWith('s', '999-0')
+    sup.stop()
+  })
+
+  it('compares by sequence when ms part is equal', async () => {
+    const redis = makeRedis(
+      [
+        { name: 'g1', pending: 1, lastDeliveredId: '5-9', lag: 1, consumers: 1 },
+        { name: 'g2', pending: 1, lastDeliveredId: '5-2', lag: 1, consumers: 1 },
+      ],
+      { g1: '5-9', g2: '5-2' },
+    )
+    const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })
+    sup.start()
+    await vi.advanceTimersByTimeAsync(100)
+    await flushPromises()
+    expect(redis.xtrim).toHaveBeenCalledWith('s', '5-2')
+    sup.stop()
+  })
+
+  it('skips trim when pending groups exist but XPENDING returns null (race)', async () => {
+    // pending>0 в XINFO, но к моменту XPENDING всё подтверждено → null → не триммим
+    const redis = makeRedis(
+      [{ name: 'g1', pending: 3, lastDeliveredId: '100-0', lag: 3, consumers: 1 }],
+      { g1: null },
+    )
+    const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })
+    sup.start()
+    await vi.advanceTimersByTimeAsync(100)
+    await flushPromises()
+    expect(redis.xtrim).not.toHaveBeenCalled()
     sup.stop()
   })
 
@@ -173,9 +241,10 @@ describe('XtrimSupervisor — trim logic', () => {
   })
 
   it('silently swallows errors from xtrim itself', async () => {
-    const redis = makeRedis([
-      { name: 'g1', pending: 1, lastDeliveredId: '100-0', lag: 1, consumers: 1 },
-    ])
+    const redis = makeRedis(
+      [{ name: 'g1', pending: 1, lastDeliveredId: '100-0', lag: 1, consumers: 1 }],
+      { g1: '95-0' },
+    )
     ;(redis.xtrim as ReturnType<typeof vi.fn>).mockRejectedValue(new Error('xtrim failed'))
 
     const sup = new XtrimSupervisor({ redis, stream: 's', intervalMs: 100 })

packages/ship-reader/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@coopenomics/coopos-ship-reader",
-  "version": "0.3.0",
+  "version": "0.3.1",
   "description": "Clean-room SHiP WebSocket client for EOSIO/Antelope blockchains",
   "license": "MIT",
   "author": "Coopenomics contributors",

packages/ship-reader/src/BlockStream.ts

@@ -79,7 +79,6 @@ export async function* createBlockStream(
   }
 
   let blockNum = opts.startBlock
-  let blockTime = new Date().toISOString()
 
   while (true) {
     const msg = await nextMessage()
@@ -91,7 +90,7 @@ export async function* createBlockStream(
     const [type, raw] = decodeResult(new Uint8Array(msg), abi)
     if (type !== 'get_blocks_result_v0') continue
 
-    const block = decodeBlocksResult(raw, abi, blockNum, '', blockTime)
+    const block = decodeBlocksResult(raw, abi, blockNum, '')
     blockNum = block.thisBlock.blockNum
 
     yield block

packages/ship-reader/src/ShipProtocol.ts

@@ -43,6 +43,7 @@ interface RawBlocksResult {
   last_irreversible: RawBlockPos
   this_block: RawBlockPos | null
   prev_block: RawBlockPos | null
+  block: Bytes | null
   traces: Bytes | null
   deltas: Bytes | null
 }
@@ -113,7 +114,7 @@ export function decodeStatusResult(raw: unknown): { chainId: string; head: Block
   }
 }
 
-export function decodeBlocksResult(raw: unknown, abi: ShipAbi, blockNum: number, blockId: string, blockTime: string): ShipBlock {
+export function decodeBlocksResult(raw: unknown, abi: ShipAbi, blockNum: number, blockId: string): ShipBlock {
   const r = raw as RawBlocksResult
 
   // ABI decoder возвращает checksum256 как Checksum256 объект (не строку) —
@@ -127,6 +128,22 @@ export function decodeBlocksResult(raw: unknown, abi: ShipAbi, blockNum: number,
     ? { blockNum: r.prev_block.block_num, blockId: String(r.prev_block.block_id) }
     : null
 
+  // Время блока берём из on-chain header (signed_block.timestamp), а НЕ из wall-clock.
+  // r.block — optional<bytes> с сериализованным signed_block, его база — block_header,
+  // первое поле которого timestamp. Декодим как block_header: читаются только
+  // header-поля, тело блока (transactions) не десериализуем.
+  let blockTime = ''
+  if (r.block && r.block.length > 0) {
+    try {
+      const headerDecoded = Serializer.decode({ data: r.block.array, type: 'block_header', abi })
+      const header = Serializer.objectify(headerDecoded) as { timestamp?: string }
+      blockTime = header.timestamp ? String(header.timestamp) : ''
+    } catch {
+      // Не смогли распарсить header — оставляем пустым, downstream залогирует.
+      blockTime = ''
+    }
+  }
+
   const txTraces = decodeVector<[string, RawTransactionTrace]>(r.traces, 'transaction_trace', abi)
   const tableDeltaVariants = decodeVector<[string, RawTableDelta]>(r.deltas, 'table_delta', abi)
 
@@ -228,5 +245,5 @@ export function decodeBlocksResult(raw: unknown, abi: ShipAbi, blockNum: number,
     }
   }
 
-  return { thisBlock, head, lastIrreversible, prevBlock, traces, deltas }
+  return { thisBlock, head, lastIrreversible, prevBlock, blockTime, traces, deltas }
 }

packages/ship-reader/src/types/ship.ts

@@ -62,6 +62,8 @@ export interface ShipBlock {
   readonly head: BlockPosition
   readonly lastIrreversible: BlockPosition
   readonly prevBlock: BlockPosition | null
+  /** On-chain время блока (signed_block.timestamp), ISO-строка. '' если block не запрошен. */
+  readonly blockTime: string
   readonly traces: readonly ShipTrace[]
   readonly deltas: readonly ShipDelta[]
 }

packages/ship-reader/test/unit/shipProtocol.test.ts

@@ -123,6 +123,13 @@ function makeShipAbi(): ABI {
         { name: 'name', type: 'string' },
         { name: 'rows', type: 'row[]' },
       ] },
+      // Упрощённый block_header: реальный signed_block начинается с этих байт,
+      // decodeBlocksResult декодит r.block как 'block_header' и читает timestamp.
+      // Достаточно первых двух полей — self-consistent encode/decode тем же ABI.
+      { name: 'block_header', base: '', fields: [
+        { name: 'timestamp', type: 'block_timestamp_type' },
+        { name: 'producer', type: 'name' },
+      ] },
     ],
     actions: [],
     tables: [],
@@ -186,7 +193,7 @@ describe('ShipProtocol — decodeBlocksResult (no traces / no deltas)', () => {
       traces: null,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 100, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 100, 'c'.repeat(64))
     expect(block.traces).toHaveLength(0)
     expect(block.deltas).toHaveLength(0)
     expect(block.thisBlock.blockNum).toBe(100)
@@ -202,7 +209,7 @@ describe('ShipProtocol — decodeBlocksResult (no traces / no deltas)', () => {
       traces: null,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 42, 'fallback-id', '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 42, 'fallback-id')
     expect(block.thisBlock.blockNum).toBe(42)
     expect(block.thisBlock.blockId).toBe('fallback-id')
     expect(block.prevBlock).toBeNull()
@@ -219,7 +226,7 @@ describe('ShipProtocol — decodeBlocksResult (no traces / no deltas)', () => {
       traces: null,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 100, 'fallback', '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 100, 'fallback')
     expect(typeof block.thisBlock.blockId).toBe('string')
     expect(typeof block.head.blockId).toBe('string')
     expect(typeof block.lastIrreversible.blockId).toBe('string')
@@ -316,7 +323,7 @@ describe('ShipProtocol — decodeBlocksResult (real decoded traces)', () => {
       deltas: null,
     }
 
-    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
     expect(block.traces).toHaveLength(1)
 
     const t = block.traces[0]!
@@ -346,7 +353,7 @@ describe('ShipProtocol — decodeBlocksResult (real decoded traces)', () => {
       traces: tracesBytes,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
     const r = block.traces[0]!.receipt!
     expect(typeof r.receiver).toBe('string')
     expect(typeof r.actDigest).toBe('string')
@@ -367,7 +374,7 @@ describe('ShipProtocol — decodeBlocksResult (real decoded traces)', () => {
       traces: tracesBytes,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
     expect(block.traces[0]!.globalSequence).toBe(42n)
   })
 
@@ -382,7 +389,7 @@ describe('ShipProtocol — decodeBlocksResult (real decoded traces)', () => {
       traces: tracesBytes,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
     expect(block.traces[0]!.receipt).toBeNull()
     // Без receipt и top-level global_sequence — fallback на 0n
     expect(block.traces[0]!.globalSequence).toBe(0n)
@@ -399,7 +406,7 @@ describe('ShipProtocol — decodeBlocksResult (real decoded traces)', () => {
       traces: tracesBytes,
       deltas: null,
     }
-    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
     expect(block.traces[0]!.actRaw).toBeInstanceOf(Uint8Array)
   })
 })
@@ -435,7 +442,7 @@ describe('ShipProtocol — decodeBlocksResult (deltas)', () => {
       traces: null,
       deltas: tableDelta,
     }
-    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64))
     expect(block.deltas).toHaveLength(1)
     const d = block.deltas[0]!
     expect(d.name).toBe('contract_row')
@@ -470,7 +477,7 @@ describe('ShipProtocol — decodeBlocksResult (deltas)', () => {
       traces: null,
       deltas: tableDelta,
     }
-    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64))
     expect(block.deltas).toHaveLength(1)
     const d = block.deltas[0]!
     expect(d.name).toBe('permission')
@@ -499,12 +506,51 @@ describe('ShipProtocol — decodeBlocksResult (deltas)', () => {
       traces: null,
       deltas: tableDelta,
     }
-    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64), '2024-01-01T00:00:00.000')
+    const block = decodeBlocksResult(raw, abi, 300, 'c'.repeat(64))
     // Некорректная row пропущена без throw
     expect(block.deltas).toHaveLength(0)
   })
 })
 
+describe('ShipProtocol — decodeBlocksResult (block_time из header)', () => {
+  // Регресс-гард на баг #2: block_time брался из wall-clock (new Date()),
+  // одинаковый для всех блоков. Теперь — из on-chain signed_block.timestamp.
+  it('extracts on-chain block_time from signed_block header (not wall-clock)', () => {
+    const abi = makeShipAbi()
+    const headerBytes = Serializer.encode({
+      object: { timestamp: '2026-05-31T18:01:22.000', producer: 'eosio' },
+      type: 'block_header',
+      abi,
+    })
+    const raw = {
+      head: { block_num: 200, block_id: 'a'.repeat(64) },
+      last_irreversible: { block_num: 190, block_id: 'b'.repeat(64) },
+      this_block: { block_num: 200, block_id: 'c'.repeat(64) },
+      prev_block: null,
+      block: headerBytes,
+      traces: null,
+      deltas: null,
+    }
+    const block = decodeBlocksResult(raw, abi, 200, 'c'.repeat(64))
+    expect(block.blockTime).toContain('2026-05-31T18:01:22')
+  })
+
+  it('block_time is empty when block not fetched (fetch_block disabled)', () => {
+    const abi = makeShipAbi()
+    const raw = {
+      head: { block_num: 100, block_id: 'a'.repeat(64) },
+      last_irreversible: { block_num: 90, block_id: 'b'.repeat(64) },
+      this_block: { block_num: 100, block_id: 'c'.repeat(64) },
+      prev_block: null,
+      block: null,
+      traces: null,
+      deltas: null,
+    }
+    const block = decodeBlocksResult(raw, abi, 100, 'c'.repeat(64))
+    expect(block.blockTime).toBe('')
+  })
+})
+
 describe('ShipProtocol — defence against wharfkit returning typed objects', () => {
   // Регрессионный тест: если кто-то удалит String() обёртки — эти тесты упадут.
   it('all trace string fields survive JSON.stringify roundtrip without [object Object]', () => {