FABN-1612 cherrypick from master to release-2.2 (#305)

* FABN-1608 remove error message from discovery (#289)

When a group has insufficient members after filtering
based on user's requirements (specific mspids, specific peers,
ledger height) an error message was posted. This will be
changed to a debug message as this is not an error, but an expected
result of the filtering.

Signed-off-by: Bret Harrison <[email protected]>

* FABN-1607 Allow future startBlocks on EventService (#286)

Allow the setting of startBlocks when setting up an EventService
that have not happened and may not happen for sometime. We must
keep the stream setup timer to be able to monitor for errors
from the Peer event service, however we will check if the timer
pops if we are waiting for a defined start block. In this case
we will assume that the stream is all good and resolve the start
service for the user call.

Signed-off-by: Bret Harrison <[email protected]>

* FABN-1560 add disconnected peer to channel (#288)

Allow for all service endpoints to be disconnected when added
to a channel. When a service endpoint (peer, events, orderer, discovery) is
added to a channel, it may not be active or it may go down between usages.
The service connection will be checked and reset when an outbound request is made.

Signed-off-by: Bret Harrison <[email protected]>

* FABN-1601 discovery required organizations (#291)

When a discovery user knows the organizations that are needed
to endorse a proposal, they may use the 'requiredOrgs' setting
to have the DiscoveryHandler send it to peer that have been
discovered for the required organizations.

Signed-off-by: Bret Harrison <[email protected]>

* FABN-1596 add noPrivateReads (#293)

Allow users to set the "noPrivateReads" when setting up a
discovery hint that will be used as the interest when the
discovery service builds an endorsement plan.

Signed-off-by: Bret Harrison <[email protected]>

* FABN-1611 Blockdecoder needs kv_reads array (#297)

The RangeQueryInfo raw_reads attribute is a QueryRead message object
which has an array of KVRead objects and is not an array itself.

Signed-off-by: Bret Harrison <[email protected]>

Author: harrisob

fabric-common/lib/BlockDecoder.js

@@ -1513,7 +1513,7 @@ function decodeRangeQueryInfo(rangeQueryInfoProto) {
 	if (rangeQueryInfoProto.raw_reads) {
 		range_query_info.raw_reads = {};
 		range_query_info.raw_reads.kv_reads = [];
-		for (const kVReadProto of rangeQueryInfoProto.raw_reads) {
+		for (const kVReadProto of rangeQueryInfoProto.raw_reads.kv_reads) {
 			range_query_info.raw_reads.kv_reads.push(decodeKVRead(kVReadProto));
 		}
 	} else if (rangeQueryInfoProto.reads_merkle_hashes) {

fabric-common/lib/Channel.js

@@ -255,8 +255,8 @@ const Channel = class {
 		if (!(endorser.type === 'Endorser')) {
 			throw Error('Missing valid endorser instance');
 		}
-		if (!(endorser.connected)) {
-			throw Error('Endorser must be connected');
+		if (!endorser.isConnectable()) {
+			throw Error('Endorser must be connectable');
 		}
 		const name = endorser.name;
 		const check = this.endorsers.get(name);
@@ -336,8 +336,8 @@ const Channel = class {
 		if (!(committer.type === 'Committer')) {
 			throw Error('Missing valid committer instance');
 		}
-		if (!(committer.connected)) {
-			throw Error('Committer must be connected');
+		if (!committer.isConnectable()) {
+			throw Error('Committer must be connectable');
 		}
 		const name = committer.name;
 		const check = this.committers.get(name);

fabric-common/lib/Commit.js

@@ -182,19 +182,28 @@ class Commit extends Proposal {
 		} else if (targets) {
 			logger.debug('%s - sending to the targets', method);
 			const committers = this.channel.getTargetCommitters(targets);
-			let bad_result = {};
-			bad_result.status = 'UNKNOWN';
+			let result;
 			for (const committer of committers) {
-				const result = await committer.sendBroadcast(envelope, requestTimeout);
-				if (result.status === 'SUCCESS') {
-
-					return result;
+				const isConnected = await committer.checkConnection();
+				if (isConnected) {
+					try {
+						result = await committer.sendBroadcast(envelope, requestTimeout);
+						if (result.status === 'SUCCESS') {
+							break;
+						}
+					} catch (error) {
+						logger.error('%s - Unable to commit on %s ::%s', method, committer.name, error);
+						result = error;
+					}
 				} else {
-					bad_result = result;
+					result = new Error(`Committer ${committer.name} is not connected`);
 				}
 			}
+			if (result instanceof Error) {
+				throw result;
+			}
 
-			return bad_result;
+			return result;
 		} else {
 			throw checkParameter('targets');
 		}

fabric-common/lib/DiscoveryHandler.js

@@ -112,18 +112,23 @@ class DiscoveryHandler extends ServiceHandler {
 			for (const committer of committers) {
 				logger.debug('%s - sending to committer %s', method, committer.name);
 				try {
-					const results = await committer.sendBroadcast(signedEnvelope, timeout);
-					if (results) {
-						if (results.status === 'SUCCESS') {
-							logger.debug('%s - Successfully sent transaction to the committer %s', method, committer.name);
-							return results;
+					const isConnected = await committer.checkConnection();
+					if (isConnected) {
+						const results = await committer.sendBroadcast(signedEnvelope, timeout);
+						if (results) {
+							if (results.status === 'SUCCESS') {
+								logger.debug('%s - Successfully sent transaction to the committer %s', method, committer.name);
+								return results;
+							} else {
+								logger.debug('%s - Failed to send transaction successfully to the committer status:%s', method, results.status);
+								return_error = new Error('Failed to send transaction successfully to the committer status:' + results.status);
+							}
 						} else {
-							logger.debug('%s - Failed to send transaction successfully to the committer status:%s', method, results.status);
-							return_error = new Error('Failed to send transaction successfully to the committer status:' + results.status);
+							return_error = new Error('Failed to send transaction to the committer');
+							logger.debug('%s - Failed to send transaction to the committer %s', method, committer.name);
 						}
 					} else {
-						return_error = new Error('Failed to send transaction to the committer');
-						logger.debug('%s - Failed to send transaction to the committer %s', method, committer.name);
+						return_error = new Error(`Committer ${committer.name} is not connected`);
 					}
 				} catch (error) {
 					logger.debug('%s - Caught: %s', method, error.toString());
@@ -155,7 +160,20 @@ class DiscoveryHandler extends ServiceHandler {
 
 		const results = await this.discovery.getDiscoveryResults(true);
 
-		if (results && results.endorsement_plan) {
+		if (results && request.requiredOrgs) {
+			// special case when user knows which organizations to send the endorsement
+			// let's build our own endorsement plan so that we can use the sorting and sending code
+			const endorsement_plan = this._buildRequiredOrgPlan(results.peers_by_org);
+
+			// remove all org and peer
+			const orgs_request = {
+				sort: request.sort,
+				preferredHeightGap: request.preferredHeightGap
+			};
+
+			return this._endorse(endorsement_plan, orgs_request, signedProposal, timeout);
+		} else if (results && results.endorsement_plan) {
+			// normal processing of the discovery results
 			const working_discovery = JSON.parse(JSON.stringify(results.endorsement_plan));
 
 			return this._endorse(working_discovery, request, signedProposal, timeout);
@@ -259,7 +277,7 @@ class DiscoveryHandler extends ServiceHandler {
 			if (required > group.peers.length) {
 				results.success = false;
 				const error = new Error(`Endorsement plan group does not contain enough peers (${group.peers.length}) to satisfy policy (required:${required})`);
-				logger.error(error);
+				logger.debug(error.message);
 				results.endorsements.push(error);
 				break; // no need to look at other groups, this layout failed
 			}
@@ -302,6 +320,23 @@ class DiscoveryHandler extends ServiceHandler {
 		return responses;
 	}
 
+	_buildRequiredOrgPlan(peers_by_org) {
+		const method = '_buildRequiredOrgPlan';
+		logger.debug('%s - starting', method);
+		const endorsement_plan = {plan_id: 'required organizations'};
+		endorsement_plan.groups = {};
+		endorsement_plan.layouts = [{}]; // only one layout which will have all organizations
+
+		for (const mspid in peers_by_org) {
+			logger.debug(`${method} - found org:${mspid}`);
+			endorsement_plan.groups[mspid] = {}; // make a group for each organization
+			endorsement_plan.groups[mspid].peers = peers_by_org[mspid].peers; // now put in all peers from that organization
+			endorsement_plan.layouts[0][mspid] = 1; // add this org to the one layout and require one peer to endorse
+		}
+
+		return endorsement_plan;
+	}
+
 	/*
 	 * utility method to build a promise that will return one of the required
 	 * endorsements or an error object
@@ -326,9 +361,14 @@ class DiscoveryHandler extends ServiceHandler {
 							logger.debug('%s - send endorsement to %s', method, peer_info.name);
 							peer_info.in_use = true;
 							try {
-								endorsement = await peer.sendProposal(proposal, timeout);
-								// save this endorsement results in case we try this peer again
-								logger.debug('%s - endorsement completed to %s', method, peer_info.name);
+								const isConnected = await peer.checkConnection();
+								if (isConnected) {
+									endorsement = await peer.sendProposal(proposal, timeout);
+									// save this endorsement results in case we try this peer again
+									logger.debug('%s - endorsement completed to %s', method, peer_info.name);
+								} else {
+									endorsement = new Error(`Peer ${peer.name} is not connected`);
+								}
 							} catch (error) {
 								endorsement = error;
 								logger.error('%s - error on endorsement to %s error %s', method, peer_info.name, error);

fabric-common/lib/DiscoveryService.js

@@ -70,10 +70,10 @@ class DiscoveryService extends ServiceAction {
 		}
 
 		for (const discoverer of targets) {
-			if (discoverer.connected || discoverer.isConnectable()) {
-				logger.debug('%s - target is or could be connected %s', method, discoverer.name);
+			if (discoverer.isConnectable()) {
+				logger.debug('%s - target is connectable%s', method, discoverer.name);
 			} else {
-				throw Error(`Discoverer ${discoverer.name} is not connected`);
+				throw Error(`Discoverer ${discoverer.name} is not connectable`);
 			}
 		}
 		// must be all targets are connected
@@ -103,14 +103,18 @@ class DiscoveryService extends ServiceAction {
 	 *  sending to the peer.
 	 * @property {Endorsement} [endorsement] - Optional. Include the endorsement
 	 *  instance to build the discovery request based on the proposal.
-	 *  This will get the discovery interest (chaincode names and collections)
+	 *  This will get the discovery interest (chaincode names, collections and "no private reads")
 	 *  from the endorsement instance. Use the {@link Proposal#addCollectionInterest}
-	 *  to add collections to the endorsement's chaincode. Use the
-	 *  {@link Proposal#addChaincodeCollectionsInterest} to add chaincodes
-	 *  and collections that will be called by the endorsement's chaincode.
+	 *  to add collections to the endorsement's chaincode.
+	 *  Use the {@link Proposal#setNoPrivateReads} to set the proposals "no private reads"
+	 *  setting of the discovery interest.
+	 *  Use the {@link Proposal#addCollectionInterest} to add chaincodes,
+	 *  collections, and no private reads that will be used to get an endorsement plan
+	 *  from the peer's discovery service.
 	 * @property {DiscoveryChaincode} [interest] - Optional. An
-	 *  array of {@link DiscoveryChaincodeInterest} that have chaincodes
-	 *  and collections to calculate the endorsement plans.
+	 *  array of {@link DiscoveryChaincodeInterest} that have chaincodes, collections,
+	 *  and "no private reads" to help the peer's discovery service calculate the
+	 *  endorsement plan.
 	 * @example <caption>"single chaincode"</caption>
 	 *  [
 	 *     { name: "mychaincode"}
@@ -121,17 +125,21 @@ class DiscoveryService extends ServiceAction {
 	 *  ]
 	 * @example <caption>"single chaincode with a collection"</caption>
 	 *  [
-	 *     { name: "mychaincode", collection_names: ["mycollection"] }
+	 *     { name: "mychaincode", collectionNames: ["mycollection"] }
+	 *  ]
+	 * @example <caption>"single chaincode with a collection allowing no private data reads"</caption>
+	 *  [
+	 *     { name: "mychaincode", collectionNames: ["mycollection"], noPrivateReads: true }
 	 *  ]
 	 * @example <caption>"chaincode to chaincode with a collection"</caption>
 	 *  [
-	 *     { name: "mychaincode", collection_names: ["mycollection"] },
-	 *     { name: "myotherchaincode", collection_names: ["mycollection"] }}
+	 *     { name: "mychaincode", collectionNames: ["mycollection"] },
+	 *     { name: "myotherchaincode", collectionNames: ["mycollection"] }}
 	 *  ]
 	 * @example <caption>"chaincode to chaincode with collections"</caption>
 	 *  [
-	 *     { name: "mychaincode", collection_names: ["mycollection", "myothercollection"] },
-	 *     { name: "myotherchaincode", collection_names: ["mycollection", "myothercollection"] }}
+	 *     { name: "mychaincode", collectionNames: ["mycollection", "myothercollection"] },
+	 *     { name: "myotherchaincode", collectionNames: ["mycollection", "myothercollection"] }}
 	 *  ]
 	 */
 
@@ -144,7 +152,8 @@ class DiscoveryService extends ServiceAction {
 	/**
 	 * @typedef {Object} DiscoveryChaincodeCall
 	 * @property {string} name - The name of the chaincode
-	 * @property {string[]} [collection_names] - The names of the related collections
+	 * @property {string[]} [collectionNames] - The names of the related collections
+	 * @property {boolean} [noPrivateReads] - Indicates we do not need to read from private data
 	 */
 
 	/**
@@ -195,7 +204,7 @@ class DiscoveryService extends ServiceAction {
 			queries.push(localQuery);
 		}
 
-		// add a chaincode query to get endorsement plans
+		// add a discovery chaincode query to get endorsement plans
 		if (endorsement || interest) {
 			const interests = [];
 
@@ -285,9 +294,12 @@ class DiscoveryService extends ServiceAction {
 		for (const target of this.targets) {
 			logger.debug(`${method} - about to discover on ${target.endpoint.url}`);
 			try {
-				response = await target.sendDiscovery(signedEnvelope, this.requestTimeout);
-				this.currentTarget = target;
-				break;
+				const isConnected = await target.checkConnection();
+				if (isConnected) {
+					response = await target.sendDiscovery(signedEnvelope, this.requestTimeout);
+					this.currentTarget = target;
+					break;
+				}
 			} catch (error) {
 				response = error;
 			}
@@ -376,6 +388,9 @@ class DiscoveryService extends ServiceAction {
 			const chaincodeCall = fabproto6.discovery.ChaincodeCall.create();
 			if (typeof chaincode.name === 'string') {
 				chaincodeCall.name = chaincode.name;
+				if (chaincode.noPrivateReads) {
+					chaincodeCall.no_private_reads = chaincode.noPrivateReads;
+				}
 				// support both names
 				if (chaincode.collection_names) {
 					_getCollectionNames(chaincode.collection_names, chaincodeCall);
@@ -738,7 +753,7 @@ class DiscoveryService extends ServiceAction {
 	}
 }
 
-function _getCollectionNames(names, chaincode_call) {
+function _getCollectionNames(names, chaincodeCall) {
 	if (Array.isArray(names)) {
 		const collection_names = [];
 		names.map(name => {
@@ -748,7 +763,9 @@ function _getCollectionNames(names, chaincode_call) {
 				throw Error('The collection name must be a string');
 			}
 		});
-		chaincode_call.collection_names = collection_names;
+		// this collection_names must be in snake case as it will
+		// be used by the gRPC create message
+		chaincodeCall.collection_names = collection_names;
 	} else {
 		throw Error('Collection names must be an array of strings');
 	}

fabric-common/lib/EventService.js

@@ -89,6 +89,7 @@ class EventService extends ServiceAction {
 		// will be set during the .build call
 		this.blockType = FILTERED_BLOCK;
 		this.replay = false;
+		this.startSpecified = false;
 
 		this.myNumber = count++;
 	}
@@ -113,13 +114,13 @@ class EventService extends ServiceAction {
 		}
 
 		for (const eventer of targets) {
-			if (eventer.connected || eventer.isConnectable()) {
-				logger.debug('%s - target is or could be connected %s', method, eventer.name);
+			if (eventer.isConnectable()) {
+				logger.debug('%s - target is connectable %s', method, eventer.name);
 			} else {
 				throw Error(`Eventer ${eventer.name} is not connectable`);
 			}
 		}
-		// must be all targets are connected
+		// must be all targets are connectable
 		this.targets = targets;
 
 		return this;
@@ -266,6 +267,7 @@ class EventService extends ServiceAction {
 				number: this.startBlock
 			});
 			this.replay = true;
+			this.startSpecified = true;
 		}
 
 		// build stop proto
@@ -356,10 +358,6 @@ class EventService extends ServiceAction {
 					logger.debug('%s - target has a stream, is already listening %s', method, target.toString());
 					startError = Error(`Event service ${target.name} is currently listening`);
 				} else {
-					if (target.isConnectable()) {
-						logger.debug('%s - target needs to connect %s', method, target.toString());
-						await target.connect(); // target endpoint has been previously assigned, but not connected yet
-					}
 					const isConnected = await target.checkConnection();
 					if (!isConnected) {
 						startError = Error(`Event service ${target.name} is not connected`);
@@ -409,8 +407,19 @@ class EventService extends ServiceAction {
 
 			logger.debug('%s - create stream setup timeout', method);
 			const connectionSetupTimeout = setTimeout(() => {
-				logger.error(`EventService[${this.name}] timed out after:${requestTimeout}`);
-				reject(Error('Event service timed out - Unable to start listening'));
+				// this service may be waiting for a start block that has not happened
+				if (this.startSpecified) {
+					logger.debug(`EventService[${this.name}] timed out after:${requestTimeout}`);
+					logger.debug(`EventService[${this.name}] not stopping service, wait indefinitely`);
+					// resolve the promise as if we did get a good response from the peer, since we did
+					// not get an "end" or "error" back indicating that the request was invalid
+					// application should have a timer just in case this peer never gets this block
+					resolve(eventer);
+				} else {
+					logger.error(`EventService[${this.name}] timed out after:${requestTimeout}`);
+					reject(Error('Event service timed out - Unable to start listening'));
+				}
+
 			}, requestTimeout);
 
 			logger.debug('%s - create stream based on blockType', method, this.blockType);

fabric-common/lib/Eventer.js

@@ -83,6 +83,8 @@ class Eventer extends ServiceEndpoint {
 		const method = `checkConnection[${this.name}:${this.myCount}]`;
 		logger.debug(`${method} - start`);
 
+		super.checkConnection();
+
 		let result = false;
 		if (this.service) {
 			try {