FABN-1607: Start event replay from previous block (#299) (#303)

Ensure that a block is received quickly to avoid the start of the event service blocking waiting for events, which in turn would block the call to Network.addBlockListener()

Signed-off-by: Mark S. Lewis <[email protected]>

Author: bestbeforetoday

fabric-network/src/impl/event/blockeventsource.ts

@@ -14,12 +14,25 @@ import { newFilteredBlockEvent } from './filteredblockeventfactory';
 import { newFullBlockEvent } from './fullblockeventfactory';
 import { OrderedBlockQueue } from './orderedblockqueue';
 import { newPrivateBlockEvent } from './privateblockeventfactory';
+import { notNullish } from '../gatewayutils';
 import Long = require('long');
 
 const logger = Logger.getLogger('BlockEventSource');
 
 const defaultBlockType: EventType = 'filtered';
 
+function newBlockQueue(options: ListenerOptions): OrderedBlockQueue {
+	const startBlock = asLong(options.startBlock);
+	return new OrderedBlockQueue(startBlock);
+}
+
+function asLong(value?: string | number | Long): Long | undefined {
+	if (notNullish(value)) {
+		return Long.fromValue(value);
+	}
+	return undefined;
+}
+
 export class BlockEventSource {
 	private readonly eventServiceManager: EventServiceManager;
 	private eventService?: EventService;
@@ -32,7 +45,7 @@ export class BlockEventSource {
 
 	constructor(eventServiceManager: EventServiceManager, options: ListenerOptions = {}) {
 		this.eventServiceManager = eventServiceManager;
-		this.blockQueue = this.newBlockQueue(options);
+		this.blockQueue = newBlockQueue(options);
 		this.asyncNotifier = new AsyncNotifier(
 			this.blockQueue.getNextBlock.bind(this.blockQueue),
 			this.notifyListeners.bind(this)
@@ -56,11 +69,6 @@ export class BlockEventSource {
 		this.started = false;
 	}
 
-	private newBlockQueue(options: ListenerOptions): OrderedBlockQueue {
-		const startBlock = options.startBlock ? Long.fromValue(options.startBlock) : undefined;
-		return new OrderedBlockQueue(startBlock);
-	}
-
 	private async start() {
 		if (this.started) {
 			return;
@@ -96,9 +104,17 @@ export class BlockEventSource {
 	}
 
 	private async startEventService() {
+		let startBlock = this.getNextBlockNumber();
+		if (startBlock) {
+			startBlock = startBlock.subtract(Long.ONE);
+			if (startBlock.isNegative()) {
+				startBlock = Long.ZERO;
+			}
+		}
+
 		const options: StartRequestOptions = {
 			blockType: this.blockType,
-			startBlock: this.getNextBlockNumber()
+			startBlock
 		};
 		await this.eventServiceManager.startEventService(this.eventService!, options);
 	}

fabric-network/src/impl/gatewayutils.ts

@@ -64,3 +64,12 @@ export function cachedResult<T>(f: () => T): () => T {
 		return value;
 	};
 }
+
+/**
+ * Typesafe check that a value is not nullish.
+ * @private
+ * @param value Any value, including null and undefined.
+ */
+export function notNullish<T>(value?: T): value is T {
+	return value !== null && value !== undefined;
+}

fabric-network/src/network.ts

@@ -16,6 +16,7 @@ import { checkpointBlockListener } from './impl/event/listeners';
 import { addListener, ListenerSession, removeListener } from './impl/event/listenersession';
 import { SharedBlockListenerSession } from './impl/event/sharedblocklistenersession';
 import { QueryHandler } from './impl/query/queryhandler';
+import { notNullish } from './impl/gatewayutils';
 import * as Logger from './logger';
 
 const logger = Logger.getLogger('Network');
@@ -355,7 +356,7 @@ export class NetworkImpl implements Network {
 			listener = checkpointBlockListener(listener, options.checkpointer);
 		}
 
-		if (options.startBlock) {
+		if (notNullish(options.startBlock)) {
 			return this.newIsolatedBlockListenerSession(listener, options);
 		} else {
 			return this.newSharedBlockListenerSession(listener, options.type);

fabric-network/test/impl/event/blocklistener.spec.ts

@@ -263,10 +263,10 @@ describe('block listener', () => {
 
 			await startListener.completePromise;
 			sinon.assert.calledWith(stub, eventService);
-			sinon.assert.neverCalledWith(stub, sinon.match.any, sinon.match.has('startBlock', sinon.match.number));
+			sinon.assert.neverCalledWith(stub, sinon.match.any, sinon.match.has('startBlock', sinon.match.defined));
 		});
 
-		it('errors trigger reconnect of event service with next block as start block if events received', async () => {
+		it('errors trigger reconnect of event service with last received block as start block if events received', async () => {
 			await network.addBlockListener(listener, listenerOptions);
 			const startListener = testUtils.newAsyncListener<void>();
 			const stub = sinon.stub(eventServiceManager, 'startEventService').callsFake(() => startListener());
@@ -275,7 +275,28 @@ describe('block listener', () => {
 			eventService.sendError(new Error('DISCONNECT'));
 
 			await startListener.completePromise;
-			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.fromNumber(2)));
+			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.ONE));
+		});
+
+		it('listener does not receive old blocks on reconnect', async () => {
+			listener = testUtils.newAsyncListener<BlockEvent>(2);
+			const event1 = newFilteredBlockEventInfo(1);
+			const event2 = newFilteredBlockEventInfo(2);
+			const startListener = testUtils.newAsyncListener<void>(2);
+			const stub = sinon.stub(eventServiceManager, 'startEventService').callsFake(() => startListener());
+
+			await network.addBlockListener(listener, listenerOptions);
+			eventService.sendEvent(event1);
+			eventService.sendError(new Error('DISCONNECT'));
+
+			await startListener.completePromise;
+
+			eventService.sendEvent(event1);
+			eventService.sendEvent(event2);
+
+			const actual = await listener.completePromise;
+			const blockNumbers = actual.map((e) => e.blockNumber);
+			expect(blockNumbers).to.deep.equal([event1.blockNumber, event2.blockNumber]);
 		});
 
 		it('listener changing event data does not affect other listeners', async () => {
@@ -325,13 +346,22 @@ describe('block listener', () => {
 	});
 
 	describe('replay', () => {
-		it('replay listener sends start block to event service', async () => {
+		it('replay listener sends (startBlock - 1) to event service', async () => {
 			const stub = sinon.stub(eventServiceManager, 'startEventService');
 
-			listenerOptions.startBlock = 2;
+			listenerOptions.startBlock = 1;
+			await network.addBlockListener(listener, listenerOptions);
+
+			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.ZERO));
+		});
+
+		it('replay listener does not send start block less than zero to event service', async () => {
+			const stub = sinon.stub(eventServiceManager, 'startEventService');
+
+			listenerOptions.startBlock = 0;
 			await network.addBlockListener(listener, listenerOptions);
 
-			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.fromNumber(2)));
+			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.ZERO));
 		});
 
 		it('replay listener does not receive events earlier than start block', async () => {
@@ -449,27 +479,27 @@ describe('block listener', () => {
 			expect(actual.blockNumber).to.equal(event.blockNumber);
 		});
 
-		it('checkpoint listener sends block number to event service', async () => {
+		it('checkpoint listener sends (block number - 1) to event service', async () => {
 			const stub = sinon.stub(eventServiceManager, 'startEventService');
 			const checkpointer = new StubCheckpointer();
 			await checkpointer.setBlockNumber(Long.fromNumber(2));
 
 			listenerOptions.checkpointer = checkpointer;
 			await network.addBlockListener(listener, listenerOptions);
 
-			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.fromNumber(2)));
+			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.ONE));
 		});
 
 		it('checkpoint block number takes precedence over startBlock option', async () => {
 			const stub = sinon.stub(eventServiceManager, 'startEventService');
 			const checkpointer = new StubCheckpointer();
-			await checkpointer.setBlockNumber(Long.fromNumber(2));
+			await checkpointer.setBlockNumber(Long.ONE);
 
 			listenerOptions.checkpointer = checkpointer;
-			listenerOptions.startBlock = 1;
+			listenerOptions.startBlock = 10;
 			await network.addBlockListener(listener, listenerOptions);
 
-			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.fromNumber(2)));
+			sinon.assert.calledWith(stub, eventService, sinon.match.has('startBlock', Long.ZERO));
 		});
 
 		it('checkpoint listener receives events from checkpoint block number', async () => {

package.json

@@ -29,7 +29,7 @@
     "unitTest:all": "export HFC_LOGGING='{\"debug\":\"test/temp/debug.log\"}' && nyc run-s unitTest:common unitTest:ca-client unitTest:network",
     "unitTest:common": "mocha --reporter list 'fabric-common/test/**/*.js'",
     "unitTest:ca-client": "mocha --reporter list 'fabric-ca-client/test/**/*.js'",
-    "unitTest:network": "npm run compile && mocha --require ts-node/register --reporter list 'fabric-network/test/**/*.{js,ts}'",
+    "unitTest:network": "run-s compile \"unitTest -- 'fabric-network/test/**/*.{js,ts}'\"",
     "unitTest": "mocha --require ts-node/register --reporter list",
     "dockerReady": "npm run dockerClean && (cd test/fixtures/docker-compose && docker-compose -f docker-compose-tls-level-db.yaml -p node up -d && sleep 15 && docker ps -a)",
     "tapeIntegration": "./scripts/npm_scripts/runTape.sh",