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>

.github/workflows/release.yml

@@ -5,18 +5,16 @@ name: Release
 # существующим git-тегом v<version>) — тогда:
 #   1. Публикуем workspace-пакеты в npm
 #   2. Создаём git-тег v<version> и пушим его
-#   3. Создаём GitHub Release с changelog'ом (changelogithub)
-#   4. Собираем и пушим Docker-образ в docker.io (Docker Hub)
+#   3. Создаём Gitea Release с changelog'ом через Gitea API
 # Если версия не менялась — workflow no-op (без ошибки).
 #
 # Требуемые secrets в настройках репозитория:
-#   NPM_TOKEN          — npm access token с правами publish на @coopenomics/*
-#                        (Gitea Actions не выдаёт OIDC id-token, поэтому npm
-#                        trusted publishing/provenance здесь недоступны —
-#                        используем классический auth через NODE_AUTH_TOKEN)
-#   DOCKERHUB_USERNAME — Docker Hub логин с доступом к организации dicoop
-#   DOCKERHUB_TOKEN    — PAT с правами Read,Write из hub.docker.com/settings/security
-#   GITHUB_TOKEN       — выдаётся автоматически GitHub Actions
+#   NPM_TOKEN    — npm access token с правами publish на @coopenomics/*
+#                  (Gitea Actions не выдаёт OIDC id-token, поэтому npm
+#                  trusted publishing/provenance здесь недоступны —
+#                  используем классический auth через NODE_AUTH_TOKEN)
+#   GITHUB_TOKEN — выдаётся автоматически Gitea Actions; используется для
+#                  создания релиза через ${GITHUB_SERVER_URL}/api/v1/...
 
 on:
   push:
@@ -27,7 +25,7 @@ concurrency:
   cancel-in-progress: false
 
 permissions:
-  contents: write   # tag + GitHub Release
+  contents: write   # tag + Gitea Release
 
 jobs:
   release:
@@ -39,7 +37,7 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          fetch-depth: 0   # нужно для git tag проверки и changelogithub
+          fetch-depth: 0   # нужно для git tag проверки и git log changelog
       - uses: pnpm/action-setup@v4
         with: { version: '10' }
       - uses: actions/setup-node@v4
@@ -90,52 +88,29 @@ jobs:
           git tag "v${{ steps.ver.outputs.version }}"
           git push origin "v${{ steps.ver.outputs.version }}"
 
-      - name: Create GitHub Release
+      # changelogithub знает только github.com и падает на git.coopenomics.world
+      # ("Can not parse GitHub repo from url ..."). Создаём релиз напрямую через
+      # Gitea REST API: GITHUB_SERVER_URL и GITHUB_REPOSITORY проставляются
+      # раннером, GITHUB_TOKEN имеет права contents:write на тот же репо.
+      # Changelog — git log между предыдущим v-тегом и текущим.
+      - name: Create Gitea Release
         if: steps.ver.outputs.released == 'true'
-        run: npx changelogithub
+        run: |
+          VERSION="${{ steps.ver.outputs.version }}"
+          PREV=$(git tag --list 'v*' --sort=-v:refname | grep -vx "v${VERSION}" | head -1)
+          if [ -n "$PREV" ]; then
+            CHANGELOG=$(git log "${PREV}..v${VERSION}" --pretty=format:'- %s')
+          else
+            CHANGELOG=$(git log "v${VERSION}" --pretty=format:'- %s')
+          fi
+          BODY=$(jq -n \
+            --arg tag "v${VERSION}" \
+            --arg body "$CHANGELOG" \
+            '{tag_name:$tag, name:$tag, body:$body, draft:false, prerelease:false}')
+          curl -fsS -X POST \
+            -H "Authorization: token ${GITHUB_TOKEN}" \
+            -H "Content-Type: application/json" \
+            -d "$BODY" \
+            "${GITHUB_SERVER_URL}/api/v1/repos/${GITHUB_REPOSITORY}/releases"
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-
-  docker:
-    name: Build & push Docker image
-    runs-on: ubuntu-22.04
-    needs: release
-    if: needs.release.outputs.released == 'true'
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          ref: v${{ needs.release.outputs.version }}
-      - uses: pnpm/action-setup@v4
-        with: { version: '10' }
-      - uses: actions/setup-node@v4
-        with: { node-version: '20', cache: 'pnpm' }
-      - run: pnpm install --frozen-lockfile
-      - run: pnpm --filter @coopenomics/coopos-ship-reader build
-      - run: pnpm --filter @coopenomics/parser2 build
-
-      - uses: docker/setup-buildx-action@v3
-      # Docker Hub auth: требует secrets DOCKERHUB_USERNAME + DOCKERHUB_TOKEN
-      # (Access Token из hub.docker.com/settings/security с правами Read,Write)
-      - uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-      - name: Docker meta
-        id: meta
-        uses: docker/metadata-action@v5
-        with:
-          images: dicoop/parser2
-          tags: |
-            type=semver,pattern={{version}},value=v${{ needs.release.outputs.version }}
-            type=semver,pattern={{major}}.{{minor}},value=v${{ needs.release.outputs.version }}
-            type=raw,value=latest
-      - uses: docker/build-push-action@v5
-        with:
-          context: .
-          file: Dockerfile
-          push: true
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          cache-from: type=gha
-          cache-to: type=gha,mode=max
-          platforms: linux/amd64,linux/arm64

packages/parser2/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@coopenomics/parser2",
-  "version": "1.1.0",
+  "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/adapters/ShipReaderAdapter.ts

@@ -86,6 +86,6 @@ export class ShipReaderAdapter implements ChainClient {
    * содержит hardcoded-схемы нативных таблиц (permission, account, …).
    */
   deserializeNativeDelta(delta: ShipDelta): ShipNativeDeltaEvent {
-    return this.client.deserializer.deserializeNativeDelta(delta)
+    return this.client.deserializer.deserializeNativeDelta(delta, this.client.abi)
   }
 }

packages/parser2/src/core/BlockProcessor.ts

@@ -29,6 +29,7 @@ import { computeEventId } from '../events/eventId.js'
 import type { AbiBootstrapper } from '../abi/AbiBootstrapper.js'
 import type { AbiStore } from '../abi/AbiStore.js'
 import type { ChainClient } from '../ports/ChainClient.js'
+import { rootLogger, type Logger } from '../logger.js'
 
 interface BlockProcessorOptions {
   /** Идентификатор цепи — проставляется в каждое событие. */
@@ -68,6 +69,7 @@ export class BlockProcessor {
   private abiBootstrapper: AbiBootstrapper
   private abiStore: AbiStore
   private chainClient: ChainClient
+  private log: Logger
 
   constructor(opts: BlockProcessorOptions) {
     this.chainId = opts.chainId
@@ -75,6 +77,7 @@ export class BlockProcessor {
     this.abiBootstrapper = opts.abiBootstrapper
     this.abiStore = opts.abiStore
     this.chainClient = opts.chainClient
+    this.log = rootLogger.child({ component: 'BlockProcessor', chain_id: opts.chainId })
     this.queue = new PQueue({ concurrency: 1 })
   }
 
@@ -94,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) {
@@ -237,8 +245,13 @@ export class BlockProcessor {
             present: native.present,
           }
           nativeDeltaEvents.push({ ...partial, event_id: computeEventId(partial) })
-        } catch {
-          // Ошибки в отдельных нативных дельтах не должны прерывать весь блок
+        } catch (err) {
+          // Ошибки в отдельных нативных дельтах не должны прерывать весь блок,
+          // но молча терять их нельзя — иначе регрессии десериализации невидимы.
+          this.log.warn(
+            { block_num: blockNum, table: delta.name, err: err instanceof Error ? err.message : String(err) },
+            'native delta deserialize failed',
+          )
         }
       }
     }

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.2.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/NativeRowStream.ts

@@ -1,3 +1,4 @@
+import type { ABI } from '@wharfkit/antelope'
 import type { ShipDelta } from './types/ship.js'
 import type { NativeDeltaEvent } from './native-tables/index.js'
 import { isNativeTableName } from './native-tables/index.js'
@@ -10,9 +11,10 @@ export function filterNativeDeltas(deltas: readonly ShipDelta[]): ShipDelta[] {
 export function* streamNativeDeltas(
   deltas: readonly ShipDelta[],
   deserializer: WharfkitDeserializer,
+  abi: ABI,
 ): Generator<NativeDeltaEvent> {
   for (const delta of deltas) {
     if (!isNativeTableName(delta.name)) continue
-    yield deserializer.deserializeNativeDelta(delta)
+    yield deserializer.deserializeNativeDelta(delta, abi)
   }
 }

packages/ship-reader/src/ShipClient.ts

@@ -21,6 +21,15 @@ export class ShipClient {
     this.deserializer = new WharfkitDeserializer()
   }
 
+  /**
+   * SHiP-ABI (state_history), доступен после connect().
+   * Нужен для десериализации нативных дельт (их строки сериализованы этим ABI).
+   */
+  get abi(): ShipAbi {
+    if (!this.shipAbi) throw new ShipConnectionError('Call connect() first')
+    return this.shipAbi
+  }
+
   async connect(): Promise<void> {
     if (this.ws?.readyState === WebSocket.OPEN) return
 

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/deserializers/Deserializer.ts

@@ -5,6 +5,6 @@ import type { NativeDeltaEvent } from '../native-tables/index.js'
 export interface Deserializer {
   deserializeAction<T = Record<string, unknown>>(trace: ShipTrace, abi: ABI): Action<T>
   deserializeContractRow<T = Record<string, unknown>>(delta: ShipDelta, abi: ABI): Delta<T>
-  deserializeNativeDelta<T = Record<string, unknown>>(delta: ShipDelta): NativeDeltaEvent<T>
+  deserializeNativeDelta<T = Record<string, unknown>>(delta: ShipDelta, abi: ABI): NativeDeltaEvent<T>
   readonly name: 'wharfkit'
 }

packages/ship-reader/src/deserializers/WharfkitDeserializer.ts

@@ -57,13 +57,18 @@ export class WharfkitDeserializer implements Deserializer {
     }
   }
 
-  deserializeNativeDelta<T = Record<string, unknown>>(delta: ShipDelta): NativeDeltaEvent<T> {
+  deserializeNativeDelta<T = Record<string, unknown>>(delta: ShipDelta, abi: ABI): NativeDeltaEvent<T> {
     if (!isNativeTableName(delta.name)) {
       throw new UnknownNativeTableError(delta.name)
     }
     const table = delta.name as NativeTableName
     try {
-      const data = JSON.parse(Buffer.from(delta.rowRaw).toString('utf8')) as T
+      // rowRaw нативных таблиц — это ABI-сериализованные байты state_history,
+      // а НЕ JSON. Тип строки = имя таблицы: в ship-ABI это variant (*_v0).
+      const decoded = Serializer.decode({ data: delta.rowRaw, type: table, abi })
+      const objectified = Serializer.objectify(decoded as ABISerializable)
+      // variant оформлен как [typeName, row] — распаковываем саму строку.
+      const data = (Array.isArray(objectified) ? objectified[1] : objectified) as T
       const lookup_key = computeLookupKey(table, data as never)
       const present: boolean = delta.present
       return { present, table, data, lookup_key }

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/native-rows.test.ts

@@ -1,11 +1,50 @@
 import { describe, it, expect } from 'vitest'
+import { ABI, Serializer } from '@wharfkit/antelope'
 import { WharfkitDeserializer } from '../../src/deserializers/WharfkitDeserializer.js'
 import { computeLookupKey } from '../../src/native-tables/index.js'
 import { isNativeTableName, NATIVE_TABLE_NAMES } from '../../src/native-tables/types.js'
-import { UnknownNativeTableError } from '../../src/errors.js'
+import { UnknownNativeTableError, DeserializationError } from '../../src/errors.js'
 import type { ShipDelta } from '../../src/types/ship.js'
 import type { NativePermissionRow, NativePermissionLinkRow } from '../../src/native-tables/types.js'
 
+/**
+ * Мини-ABI в стиле state_history: каждая нативная таблица — это variant (*_v0).
+ * Достаточно полей, нужных computeLookupKey. Кодируем и декодируем одним ABI —
+ * схема самосогласована, повторять реальный EOSIO целиком не требуется.
+ */
+function makeShipAbi(): ABI {
+  return ABI.from({
+    version: 'eosio::abi/1.1',
+    types: [],
+    structs: [
+      { name: 'permission_v0', base: '', fields: [
+        { name: 'owner', type: 'name' },
+        { name: 'name', type: 'name' },
+      ] },
+      { name: 'account_v0', base: '', fields: [
+        { name: 'name', type: 'name' },
+        { name: 'creation_date', type: 'string' },
+        { name: 'abi', type: 'string' },
+      ] },
+    ],
+    actions: [],
+    tables: [],
+    variants: [
+      { name: 'permission', types: ['permission_v0'] },
+      { name: 'account', types: ['account_v0'] },
+    ],
+  })
+}
+
+const SHIP_ABI = makeShipAbi()
+
+/** Дельта с ABI-сериализованным rowRaw (как реально приходит из SHiP). */
+function abiDelta(table: string, obj: Record<string, unknown>, present = true): ShipDelta {
+  const encoded = Serializer.encode({ object: [`${table}_v0`, obj], type: table, abi: SHIP_ABI })
+  return { name: table, present, rowRaw: encoded.array }
+}
+
+/** Дельта с JSON-байтами — раньше «работала» из-за бага JSON.parse, теперь должна падать. */
 function jsonDelta(name: string, data: unknown, present = true): ShipDelta {
   return {
     name,
@@ -64,36 +103,38 @@ describe('computeLookupKey', () => {
 describe('WharfkitDeserializer — deserializeNativeDelta', () => {
   const deser = new WharfkitDeserializer()
 
-  it('deserializes permission delta with correct lookup_key', () => {
-    const permData: NativePermissionRow = {
-      owner: 'alice', name: 'active', parent: 'owner',
-      last_updated: '2024-01-01T00:00:00.000',
-      auth: { threshold: 1, keys: [], accounts: [], waits: [] },
-    }
-    const delta = jsonDelta('permission', permData)
-    const event = deser.deserializeNativeDelta<NativePermissionRow>(delta)
+  it('deserializes ABI-encoded permission delta with correct lookup_key', () => {
+    const delta = abiDelta('permission', { owner: 'alice', name: 'active' })
+    const event = deser.deserializeNativeDelta<NativePermissionRow>(delta, SHIP_ABI)
     expect(event.table).toBe('permission')
     expect(event.lookup_key).toBe('alice:active')
     expect(event.present).toBe(true)
-    expect(event.data.owner).toBe('alice')
+    expect(String(event.data.owner)).toBe('alice')
+  })
+
+  it('unwraps the *_v0 variant into a flat row (not a [name, row] tuple)', () => {
+    const delta = abiDelta('permission', { owner: 'alice', name: 'active' })
+    const event = deser.deserializeNativeDelta<NativePermissionRow>(delta, SHIP_ABI)
+    expect(Array.isArray(event.data)).toBe(false)
+    expect(String(event.data.name)).toBe('active')
   })
 
   it('present field is boolean (not string)', () => {
-    const delta = jsonDelta('account', { name: 'bob', creation_date: '', abi: '' }, false)
-    const event = deser.deserializeNativeDelta(delta)
+    const delta = abiDelta('account', { name: 'bob', creation_date: '2024-01-01T00:00:00.000', abi: '' }, false)
+    const event = deser.deserializeNativeDelta(delta, SHIP_ABI)
     expect(typeof event.present).toBe('boolean')
     expect(event.present).toBe(false)
   })
 
   it('throws UnknownNativeTableError for unknown table', () => {
     const delta: ShipDelta = { name: 'my_custom_table', present: true, rowRaw: new Uint8Array([]) }
-    expect(() => deser.deserializeNativeDelta(delta)).toThrow(UnknownNativeTableError)
+    expect(() => deser.deserializeNativeDelta(delta, SHIP_ABI)).toThrow(UnknownNativeTableError)
   })
 
-  it('all NATIVE_TABLE_NAMES tables deserialize without throwing (smoke test)', () => {
-    for (const table of NATIVE_TABLE_NAMES) {
-      const delta = jsonDelta(table, { name: 'test' })
-      expect(() => deser.deserializeNativeDelta(delta)).not.toThrow()
-    }
+  it('throws DeserializationError on non-ABI (JSON) bytes — no silent JSON.parse', () => {
+    // Регресс-гард на исходный баг: нативные строки декодятся ABI, а не JSON.parse.
+    // JSON-байты — не валидная ABI-сериализация, метод обязан бросить.
+    const delta = jsonDelta('permission', { owner: 'alice', name: 'active' })
+    expect(() => deser.deserializeNativeDelta(delta, SHIP_ABI)).toThrow(DeserializationError)
   })
 })

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]', () => {