refactor: share stream history delivery across transports

.github/workflows/build.yml

@@ -31,7 +31,7 @@ jobs:
       - run: npm run typecheck
       - run: npm run build
       - name: Check stream request lifecycle on Node ${{ matrix.node-version }}
-        run: node --test tests/native/stream-request-lifecycle.test.mjs
+        run: node --test tests/native/stream-request-lifecycle.test.mjs tests/native/stream-history-delivery.test.mjs
       - name: Check history/live stream filters and selectors on Node ${{ matrix.node-version }}
         run: node --test tests/native/stream-filter-parity.test.mjs
       - name: Typecheck Elasticsearch qualification runner

src/api/helpers/functions.ts

@@ -12,6 +12,8 @@ import {assertProtocolProjectionReady, assertProtocolRouteReady} from './protoco
 import {assertProtocolActionRequestReady, assertProtocolActionStreamReady} from './protocol-action-query.js';
 import {canonicalUint64} from '../../indexer/helpers/uint64.js';
 import {validateActionSequenceQuery} from './action-sequence-query.js';
+import type {StreamHistoryDelivery} from '../stream/history-delivery.js';
+import {SocketIOHistoryDelivery} from '../stream/socketio-history-delivery.js';
 
 export type BoolQuerySearchBody = estypes.SearchRequest & {
     query: estypes.QueryDslQueryContainer & {
@@ -101,6 +103,17 @@ export async function streamPastCommon<T extends keyof StreamTypeMap>(
     data: StreamTypeMap[T]["request"],
     dataKind: T,
     signal?: AbortSignal
+) {
+    return replayStreamHistory(fastify, new SocketIOHistoryDelivery(socket), requestUUID, data, dataKind, signal);
+}
+
+export async function replayStreamHistory<T extends keyof StreamTypeMap>(
+    fastify: FastifyInstance,
+    delivery: StreamHistoryDelivery,
+    requestUUID: string,
+    data: StreamTypeMap[T]["request"],
+    dataKind: T,
+    signal?: AbortSignal
 ) {
     if (dataKind === 'delta') assertProtocolProjectionReady(fastify, 'row-history');
     if (dataKind === 'action') assertProtocolActionStreamReady(fastify, data as StreamActionsRequest);
@@ -202,7 +215,8 @@ export async function streamPastCommon<T extends keyof StreamTypeMap>(
 
         // emit the first block
         if (init_response.hits.hits.length > 0) {
-            emitTraceInit(socket, requestUUID, init_response.hits.hits[0]._source.block_num, totalHits);
+            delivery.begin({requestId: requestUUID, firstBlock: init_response.hits.hits[0]._source.block_num,
+                totalResults: totalHits});
         }
 
         responseQueue.push(init_response);
@@ -218,7 +232,7 @@ export async function streamPastCommon<T extends keyof StreamTypeMap>(
                 if (rp._scroll_id) pendingScrollId = rp._scroll_id;
                 const enqueuedMessages: any[] = [];
                 activeProofScope?.dispose();
-                activeProofScope = createProtocolStreamScope([socket]);
+                activeProofScope = createProtocolStreamScope([delivery.connection]);
                 const proofScope = activeProofScope.scope;
                 if (dataKind === 'action') {
                     assertCompleteActionSearch(rp);
@@ -269,22 +283,16 @@ export async function streamPastCommon<T extends keyof StreamTypeMap>(
 
                 totalFiltered += filterCount;
 
-                if (socket.connected) {
+                if (delivery.connection.connected) {
                     if (enqueuedMessages.length > 0) {
                         const messages = dataKind === 'action'
                             ? await protocolActionsForStream(fastify, proofScope, enqueuedMessages) : enqueuedMessages;
                         try {
 
                             // Wait for 120 s
-                            const ackResponse = await withStreamCancellation(socket
-                                .timeout(MAX_SCROLL_TIME_SEC * 1000)
-                                .emitWithAck('message', {
-                                    reqUUID: requestUUID,
-                                    type: `${dataKind}_trace`,
-                                    mode: 'history',
-                                    messages,
-                                    filtered: filterCount
-                                }), signal);
+                            const ackResponse = await withStreamCancellation(delivery.deliver({
+                                requestId: requestUUID, kind: dataKind, messages, filtered: filterCount
+                            }, MAX_SCROLL_TIME_SEC * 1000), signal);
 
                             if (ackResponse.status !== true) {
                                 hLog(`${dataKind}_trace scroll TIMEOUT`);
@@ -306,7 +314,7 @@ export async function streamPastCommon<T extends keyof StreamTypeMap>(
                 }
 
                 if (getTotalValue(rp) === counter) {
-                    hLog(`${counter} past ${dataKind}s streamed to ${socket.id} (${totalFiltered} filtered)`);
+                    hLog(`${counter} past ${dataKind}s streamed to ${delivery.connection.id} (${totalFiltered} filtered)`);
                     break;
                 }
 
@@ -539,13 +547,7 @@ export function emitTraceInit(
     firstBlock: number,
     totalResults: number
 ) {
-    socket.emit('message', {
-        reqUUID: requestUUID,
-        type: 'trace_init',
-        mode: 'history',
-        first_block: firstBlock,
-        results: totalResults
-    });
+    new SocketIOHistoryDelivery(socket).begin({requestId: requestUUID, firstBlock, totalResults});
 }
 
 // export async function streamPastActions(

src/api/stream/history-delivery.ts

@@ -0,0 +1,37 @@
+/** Connection lifetime used by canonical proof work, independently of public framing. */
+export interface StreamConnection {
+    readonly id: string;
+    readonly connected: boolean;
+    once(event: 'disconnect', listener: () => void): unknown;
+    removeListener(event: 'disconnect', listener: () => void): unknown;
+}
+
+export interface HistoryStart {
+    readonly requestId: string;
+    /** First matching hit, not the request's resolved inclusive replay anchor. */
+    readonly firstBlock: number;
+    readonly totalResults: number;
+}
+
+export interface HistoryBatch {
+    readonly requestId: string;
+    readonly kind: 'action' | 'delta';
+    readonly messages: readonly unknown[];
+    readonly filtered: number;
+}
+
+/**
+ * Admission is transport flow control, never an application durability receipt.
+ * An adapter must bound its queued/in-flight work and stop waits on connection loss.
+ */
+export interface HistoryAcceptance {
+    readonly status: boolean;
+    readonly error?: unknown;
+}
+
+/** Shared history queries, filters and proofs use this seam for both transports. */
+export interface StreamHistoryDelivery {
+    readonly connection: StreamConnection;
+    begin(start: HistoryStart): void;
+    deliver(batch: HistoryBatch, timeoutMs: number): Promise<HistoryAcceptance>;
+}

src/api/stream/socketio-history-delivery.ts

@@ -0,0 +1,27 @@
+import type {Socket} from 'socket.io';
+import type {HistoryAcceptance, HistoryBatch, HistoryStart, StreamHistoryDelivery} from './history-delivery.js';
+
+/** Retains the legacy wire shape and its per-batch Socket.IO acknowledgement. */
+export class SocketIOHistoryDelivery implements StreamHistoryDelivery {
+    constructor(readonly connection: Socket) {}
+
+    begin(start: HistoryStart): void {
+        this.connection.emit('message', {
+            reqUUID: start.requestId,
+            type: 'trace_init',
+            mode: 'history',
+            first_block: start.firstBlock,
+            results: start.totalResults
+        });
+    }
+
+    deliver(batch: HistoryBatch, timeoutMs: number): Promise<HistoryAcceptance> {
+        return this.connection.timeout(timeoutMs).emitWithAck('message', {
+            reqUUID: batch.requestId,
+            type: `${batch.kind}_trace`,
+            mode: 'history',
+            messages: batch.messages,
+            filtered: batch.filtered
+        });
+    }
+}

tests/native/stream-history-delivery.test.mjs

@@ -0,0 +1,86 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {EventEmitter} from 'node:events';
+import {replayStreamHistory} from '../../build/api/helpers/functions.js';
+import {createProtocolStreamScope, withProtocolProofRequest} from '../../build/api/helpers/protocol-proof-request.js';
+
+const request = () => ({code: 'eosio.token', table: 'accounts', scope: 'alice', payer: '',
+    start_from: 100, read_until: 101, ignore_live: true,
+    filters: [{field: 'data.enabled', value: false}]});
+
+function fixture() {
+    const cleared = [];
+    const fastify = {manager: {chain: 'fixture', config: {api: {stream_scroll_batch: 2,
+        stream_scroll_limit: -1}, settings: {protocol: {provider: 'legacy'}}}}, elastic: {
+        search: async () => ({_scroll_id: 'owned-scroll', hits: {total: {value: 2, relation: 'eq'}, hits: [
+            {_source: {block_num: 100, block_id: 'a'.repeat(64), '@timestamp': '2026-10-23T00:00:00Z',
+                code: 'eosio.token', table: 'accounts', scope: 'alice',
+                primary_key: '18446744073709551615', data: {enabled: false}}},
+            {_source: {block_num: 101, block_id: 'b'.repeat(64), '@timestamp': '2026-10-23T00:00:00.500Z',
+                code: 'eosio.token', table: 'accounts', scope: 'alice', primary_key: '1', data: {enabled: true}}}
+        ]}}),
+        scroll: async () => {throw new Error('unexpected additional page');},
+        clearScroll: async value => {cleared.push(value.scroll_id);}
+    }};
+    return {fastify, cleared};
+}
+
+// This adapter exercises the shared delivery interface without Socket.IO methods or framing.
+function delivery() {
+    const connection = Object.assign(new EventEmitter(), {id: 'interface-fixture', connected: true});
+    const starts = [], batches = [];
+    return {connection, starts, batches,
+        begin(value) {starts.push(value);},
+        async deliver(value, timeoutMs) {batches.push({value, timeoutMs}); return {status: true};}};
+}
+
+test('a transport-neutral history delivery preserves filtering, metadata and exact keys', async () => {
+    const {fastify, cleared} = fixture(), peer = delivery();
+    const result = await replayStreamHistory(fastify, peer, 'request-a', request(), 'delta');
+    assert.deepEqual(peer.starts, [{requestId: 'request-a', firstBlock: 100, totalResults: 2}]);
+    assert.deepEqual(peer.batches, [{timeoutMs: 120000, value: {requestId: 'request-a', kind: 'delta', filtered: 1,
+        messages: [{block_num: 100, block_id: 'a'.repeat(64), code: 'eosio.token', table: 'accounts', scope: 'alice',
+            primary_key: '18446744073709551615', data: {enabled: false}, timestamp: '2026-10-23T00:00:00Z'}]}}]);
+    assert.equal(result.status, true);
+    assert.equal(result.counter, 2);
+    assert.equal(result.lastTransmittedBlock, 101);
+    assert.deepEqual(cleared, ['owned-scroll']);
+    assert.equal(peer.connection.listenerCount('disconnect'), 0);
+});
+
+test('cancellation during transport admission releases the owned history cursor', async () => {
+    const {fastify, cleared} = fixture(), peer = delivery(), controller = new AbortController();
+    let admit;
+    const entered = new Promise(resolve => {admit = resolve;});
+    peer.deliver = () => {admit(); return new Promise(() => {});};
+    const running = replayStreamHistory(fastify, peer, 'request-a', request(), 'delta', controller.signal);
+    await entered;
+    controller.abort(new Error('cancelled at delivery interface'));
+    await assert.rejects(running, /cancelled at delivery interface/);
+    assert.deepEqual(cleared, ['owned-scroll']);
+    assert.equal(peer.connection.listenerCount('disconnect'), 0);
+});
+
+test('a delivery rejection ends only its history operation and clears its cursor', async () => {
+    const {fastify, cleared} = fixture(), peer = delivery();
+    peer.deliver = async () => ({status: false, error: 'bounded delivery unavailable'});
+    const result = await replayStreamHistory(fastify, peer, 'request-a', request(), 'delta');
+    assert.deepEqual(result, {status: false, error: 'bounded delivery unavailable'});
+    assert.deepEqual(cleared, ['owned-scroll']);
+    assert.equal(peer.connection.listenerCount('disconnect'), 0);
+});
+
+test('a delivery connection disconnect cancels its active canonical proof operation', async () => {
+    const peer = delivery(), proof = createProtocolStreamScope([peer.connection]);
+    let proofSignal;
+    const running = withProtocolProofRequest(proof.scope, options => {
+        proofSignal = options.signal;
+        return new Promise(() => {});
+    });
+    peer.connection.connected = false;
+    peer.connection.emit('disconnect');
+    await assert.rejects(running, /ABI proof request was cancelled/);
+    assert.equal(proofSignal.aborted, true);
+    proof.dispose();
+    assert.equal(peer.connection.listenerCount('disconnect'), 0);
+});