diff --git a/apis/fabric-shim-api/types/index.d.ts b/apis/fabric-shim-api/types/index.d.ts index cf9559e7..aed80dd0 100644 --- a/apis/fabric-shim-api/types/index.d.ts +++ b/apis/fabric-shim-api/types/index.d.ts @@ -104,7 +104,10 @@ declare module 'fabric-shim-api' { getPrivateDataValidationParameter(collection: string, key: string): Promise; getPrivateDataByRange(collection: string, startKey: string, endKey: string): Promise & AsyncIterable; getPrivateDataByPartialCompositeKey(collection: string, objectType: string, attributes: string[]): Promise & AsyncIterable; - getPrivateDataQueryResult(collection: string, query: string): Promise & AsyncIterable; + getMultipleStates(keys: string[]): Promise; + getMultiplePrivateData(collection: string, keys: string[]): Promise; + startWriteBatch(): void; + finishWriteBatch(): Promise; } interface SplitCompositekey { diff --git a/libraries/fabric-shim/lib/batch.js b/libraries/fabric-shim/lib/batch.js new file mode 100644 index 00000000..be266c07 --- /dev/null +++ b/libraries/fabric-shim/lib/batch.js @@ -0,0 +1,76 @@ +/* +# Copyright IBM Corp. All Rights Reserved. +# +# SPDX-License-Identifier: Apache-2.0 +*/ + +'use strict'; + +const {peer} = require('@hyperledger/fabric-protos'); + +const DATA_KEY_TYPE = 'data'; +const METADATA_KEY_TYPE = 'metadata'; + +/** + * Collects consecutive write operations so they can be sent to the peer + * as one or more WRITE_BATCH_STATE messages. + * + * Same-key data writes (put/delete/purge) overwrite each other. + * Metadata writes are tracked separately from data writes. + */ +class WriteBatch { + constructor() { + this.writes = new Map(); + } + + putState(collection, key, value) { + this._writeData(this._record(key, value, collection, peer.WriteRecord.Type.PUT_STATE)); + } + + delState(collection, key) { + this._writeData(this._record(key, undefined, collection, peer.WriteRecord.Type.DEL_STATE)); + } + + purgeState(collection, key) { + this._writeData(this._record(key, undefined, collection, peer.WriteRecord.Type.PURGE_PRIVATE_DATA)); + } + + putStateMetadataEntry(collection, key, metakey, metadata) { + const stateMetadata = new peer.StateMetadata(); + stateMetadata.setMetakey(metakey); + stateMetadata.setValue(metadata); + + const rec = this._record(key, undefined, collection, peer.WriteRecord.Type.PUT_STATE_METADATA); + rec.setMetadata(stateMetadata); + this._writeMetadata(rec); + } + + records() { + return Array.from(this.writes.values()); + } + + _writeData(record) { + this.writes.set(this._mapKey(DATA_KEY_TYPE, record), record); + } + + _writeMetadata(record) { + this.writes.set(this._mapKey(METADATA_KEY_TYPE, record), record); + } + + _mapKey(type, record) { + return `${type}:${record.getCollection()}:${record.getKey()}`; + } + + _record(key, value, collection, type) { + const rec = new peer.WriteRecord(); + rec.setKey(key); + rec.setCollection(collection || ''); + rec.setType(type); + if (value != null) { + rec.setValue(value); + } + return rec; + } +} + +module.exports = WriteBatch; diff --git a/libraries/fabric-shim/lib/handler.js b/libraries/fabric-shim/lib/handler.js index eac09843..52268e03 100644 --- a/libraries/fabric-shim/lib/handler.js +++ b/libraries/fabric-shim/lib/handler.js @@ -26,6 +26,8 @@ const STATES = { Ready: 'ready' }; +const DEFAULT_MAX_SIZE_WRITE_BATCH = 100; + // message types const MSG_TYPE = { REGISTERED: peer.ChaincodeMessage.Type.REGISTERED, @@ -277,6 +279,8 @@ class ChaincodeMessageHandler { constructor(stream, chaincode) { this._stream = stream; this.chaincode = chaincode; + this.usePeerWriteBatch = false; + this.maxSizeWriteBatch = DEFAULT_MAX_SIZE_WRITE_BATCH; } // this is a long-running method that does not return until @@ -322,6 +326,7 @@ class ChaincodeMessageHandler { if (state === STATES.Established) { if (msg.type === MSG_TYPE.READY) { + this._applyPeerCapabilities(msg.payload); logger.info('Successfully established communication with peer node. State transferred to "ready"'); state = STATES.Ready; } else { @@ -374,6 +379,20 @@ class ChaincodeMessageHandler { handleMessage(msg, this, 'invoke'); } + _applyPeerCapabilities(payload) { + if (!payload || payload.length === 0) { + return; + } + + const params = peer.ChaincodeAdditionalParams.deserializeBinary(payload); + this.usePeerWriteBatch = params.getUseWriteBatch(); + this.maxSizeWriteBatch = params.getMaxSizeWriteBatch(); + + if (this.usePeerWriteBatch && this.maxSizeWriteBatch < DEFAULT_MAX_SIZE_WRITE_BATCH) { + this.maxSizeWriteBatch = DEFAULT_MAX_SIZE_WRITE_BATCH; + } + } + async handleGetMultipleStates(keys, channel_id, txid) { return await Promise.all(keys.map(key => this.handleGetState('', key, channel_id, txid))); } @@ -410,6 +429,29 @@ class ChaincodeMessageHandler { return await this._askPeerAndListen(msg, 'PutState'); } + async handleWriteBatch(writes, channel_id, txId) { + const batch = new peer.WriteBatchState(); + batch.setRecList(writes); + const msg = mapToChaincodeMessage({ + type: peer.ChaincodeMessage.Type.WRITE_BATCH_STATE, + payload: batch.serializeBinary(), + txid: txId, + channel_id: channel_id + }); + return await this._askPeerAndListen(msg, 'WriteBatchState'); + } + + async sendBatch(writes, channel_id, txId) { + if (!writes || writes.length === 0) { + return; + } + + const maxSize = this.maxSizeWriteBatch; + for (let i = 0; i < writes.length; i += maxSize) { + await this.handleWriteBatch(writes.slice(i, i + maxSize), channel_id, txId); + } + } + async handleDeleteState(collection, key, channel_id, txId) { const msgPb = new peer.DelState(); msgPb.setKey(key); @@ -700,6 +742,24 @@ async function handleMessage(msg, client, action) { method, resp.status)); + // Match Go: handleInit skips FinishWriteBatch when Init returns an error. + // handleTransaction always flushes, including on error status. + if (!(action === 'init' && resp.status >= Stub.RESPONSE_CODE.ERROR)) { + try { + await stub.finishWriteBatch(); + } catch (err) { + logger.error(util.format('%s Failed to send write batch: %s', loggerPrefix, err)); + nextStateMsg = mapToChaincodeMessage({ + type: peer.ChaincodeMessage.Type.ERROR, + payload: Buffer.from(err.toString()), + txid: msg.txid, + channel_id: msg.channel_id + }); + client._stream.write(nextStateMsg); + return; + } + } + const respPb = new peer.Response(); respPb.setMessage(resp.message); respPb.setStatus(resp.status); diff --git a/libraries/fabric-shim/lib/stub.js b/libraries/fabric-shim/lib/stub.js index 418369b5..b8abf97a 100644 --- a/libraries/fabric-shim/lib/stub.js +++ b/libraries/fabric-shim/lib/stub.js @@ -15,6 +15,7 @@ const {ChaincodeEvent} = require('@hyperledger/fabric-protos/lib/peer'); const Long = require('long'); const logger = require('./logger').getLogger('lib/stub.js'); +const WriteBatch = require('./batch'); const VALIDATION_PARAMETER = 'VALIDATION_PARAMETER'; @@ -143,6 +144,7 @@ class ChaincodeStub { this.handler = client; this.validationParameterMetakey = VALIDATION_PARAMETER; + this.writeBatch = null; if (signedProposalPb) { const decodedSP = { @@ -490,9 +492,16 @@ class ChaincodeStub { async putState(key, value) { // Access public data by setting the collection to empty string const collection = ''; + if (key === '') { + throw new Error('key must not be an empty string'); + } if (typeof value === 'string') { value = Buffer.from(value); } + if (this.writeBatch) { + this.writeBatch.putState(collection, key, value); + return; + } return await this.handler.handlePutState(collection, key, value, this.channel_id, this.txId); } @@ -506,6 +515,10 @@ class ChaincodeStub { async deleteState(key) { // Access public data by setting the collection to empty string const collection = ''; + if (this.writeBatch) { + this.writeBatch.delState(collection, key); + return; + } return await this.handler.handleDeleteState(collection, key, this.channel_id, this.txId); } @@ -519,6 +532,10 @@ class ChaincodeStub { async setStateValidationParameter(key, ep) { // Access public data by setting the collection to empty string const collection = ''; + if (this.writeBatch) { + this.writeBatch.putStateMetadataEntry(collection, key, this.validationParameterMetakey, ep); + return; + } return this.handler.handlePutStateMetadata(collection, key, this.validationParameterMetakey, ep, this.channel_id, this.txId); } @@ -975,21 +992,37 @@ class ChaincodeStub { } /** - * startWriteBatch indicates the beginning of a block of PutState/PutPrivateData calls - * that should be batched together. - * (Fallback behavior: no-op) + * startWriteBatch enables a mode where ledger writes are not immediately + * forwarded to the peer, but accumulate in a cache. The cache is sent in + * large batches either at the end of transaction execution or after + * finishWriteBatch is called. + * + * If write batching is not supported by the peer, this method has no effect + * and writes to the ledger continue to be processed immediately. */ startWriteBatch() { logger.debug('startWriteBatch called'); + if (this.handler.usePeerWriteBatch && !this.writeBatch) { + this.writeBatch = new WriteBatch(); + } } /** - * finishWriteBatch sends the currently accumulated batch of state writes to the peer. - * (Fallback behavior: no-op) + * finishWriteBatch sends accumulated writes in large batches to the peer + * if startWriteBatch has been called before it. + * + * If write batching is not supported by the peer or no write batch has been + * started, this method has no effect. * @async */ async finishWriteBatch() { logger.debug('finishWriteBatch called'); + try { + const writes = this.writeBatch ? this.writeBatch.records() : null; + await this.handler.sendBatch(writes, this.channel_id, this.txId); + } finally { + this.writeBatch = null; + } } /** @@ -1032,6 +1065,9 @@ class ChaincodeStub { if (!collection || typeof collection !== 'string') { throw new Error('collection must be a valid string'); } + if (key === '') { + throw new Error('key must not be an empty string'); + } if (!key || typeof key !== 'string') { throw new Error('key must be a valid string'); } @@ -1042,6 +1078,10 @@ class ChaincodeStub { value = Buffer.from(value); } + if (this.writeBatch) { + this.writeBatch.putState(collection, key, value); + return; + } return this.handler.handlePutState(collection, key, value, this.channel_id, this.txId); } @@ -1064,6 +1104,10 @@ class ChaincodeStub { if (!key || typeof key !== 'string') { throw new Error('key must be a valid string'); } + if (this.writeBatch) { + this.writeBatch.delState(collection, key); + return; + } return this.handler.handleDeleteState(collection, key, this.channel_id, this.txId); } @@ -1088,6 +1132,10 @@ class ChaincodeStub { if (!key || typeof key !== 'string') { throw new Error('key must be a valid string'); } + if (this.writeBatch) { + this.writeBatch.purgeState(collection, key); + return; + } return await this.handler.handlePurgeState(collection, key, this.channel_id, this.txId); } @@ -1101,6 +1149,10 @@ class ChaincodeStub { * @param {Buffer} ep endorsement policy */ async setPrivateDataValidationParameter(collection, key, ep) { + if (this.writeBatch) { + this.writeBatch.putStateMetadataEntry(collection, key, this.validationParameterMetakey, ep); + return; + } return this.handler.handlePutStateMetadata(collection, key, this.validationParameterMetakey, ep, this.channel_id, this.txId); } diff --git a/libraries/fabric-shim/test/typescript/chaincode.ts b/libraries/fabric-shim/test/typescript/chaincode.ts index 73ec0410..bd3b3c2b 100644 --- a/libraries/fabric-shim/test/typescript/chaincode.ts +++ b/libraries/fabric-shim/test/typescript/chaincode.ts @@ -221,6 +221,8 @@ class TestTS implements ChaincodeInterface { const mspid: string = creator.mspid; const invokeChaincode: ChaincodeResponse = await stub.invokeChaincode('ccid', ['bob', 'duck'], 'channelid'); + stub.startWriteBatch(); + await stub.finishWriteBatch(); } testClientIdentity(stub: ChaincodeStub): void { diff --git a/libraries/fabric-shim/test/unit/handler.js b/libraries/fabric-shim/test/unit/handler.js index a8c0526c..fa145757 100644 --- a/libraries/fabric-shim/test/unit/handler.js +++ b/libraries/fabric-shim/test/unit/handler.js @@ -705,6 +705,62 @@ describe('Handler', () => { expect(handleTransactionSpy.notCalled).to.be.true; }); + it('should leave write batching disabled when READY has no payload', () => { + eventReg.data(registeredMsg); + eventReg.data(establishedMsg); + + expect(handler.usePeerWriteBatch).to.equal(false); + expect(handler.maxSizeWriteBatch).to.equal(100); + }); + + it('should enable write batching from READY ChaincodeAdditionalParams', () => { + const params = new peer.ChaincodeAdditionalParams(); + params.setUseWriteBatch(true); + params.setMaxSizeWriteBatch(250); + const readyMsg = mapToChaincodeMessage({ + type: MSG_TYPE.READY, + payload: params.serializeBinary() + }); + + eventReg.data(registeredMsg); + eventReg.data(readyMsg); + + expect(handler.usePeerWriteBatch).to.equal(true); + expect(handler.maxSizeWriteBatch).to.equal(250); + }); + + it('should raise maxSizeWriteBatch to the default minimum', () => { + const params = new peer.ChaincodeAdditionalParams(); + params.setUseWriteBatch(true); + params.setMaxSizeWriteBatch(10); + const readyMsg = mapToChaincodeMessage({ + type: MSG_TYPE.READY, + payload: params.serializeBinary() + }); + + eventReg.data(registeredMsg); + eventReg.data(readyMsg); + + expect(handler.usePeerWriteBatch).to.equal(true); + expect(handler.maxSizeWriteBatch).to.equal(100); + }); + + it('should keep write batching disabled when the peer does not enable it', () => { + const params = new peer.ChaincodeAdditionalParams(); + params.setUseWriteBatch(false); + params.setMaxSizeWriteBatch(10); + const readyMsg = mapToChaincodeMessage({ + type: MSG_TYPE.READY, + payload: params.serializeBinary() + }); + + eventReg.data(registeredMsg); + eventReg.data(readyMsg); + + expect(handler.usePeerWriteBatch).to.equal(false); + expect(handler.maxSizeWriteBatch).to.equal(10); + }); + it ('should call handleMsgResponse when in state ready and MSG_TYPE equals RESPONSE', () => { eventReg.data(registeredMsg); eventReg.data(establishedMsg); @@ -1043,6 +1099,107 @@ describe('Handler', () => { }); }); + describe('handleWriteBatch', () => { + const key = 'theKey'; + const value = Buffer.from('some value'); + const collection = ''; + + let expectedMsg; + let rec; + + before(() => { + rec = new peer.WriteRecord(); + rec.setKey(key); + rec.setValue(value); + rec.setCollection(collection); + rec.setType(peer.WriteRecord.Type.PUT_STATE); + + const batch = new peer.WriteBatchState(); + batch.setRecList([rec]); + + expectedMsg = mapToChaincodeMessage({ + type: peer.ChaincodeMessage.Type.WRITE_BATCH_STATE, + payload: batch.serializeBinary(), + channel_id: 'theChannelID', + txid: 'theTxID' + }); + }); + + afterEach(() => { + Handler = rewire('../../../fabric-shim/lib/handler.js'); + sandbox.restore(); + }); + + it('should resolve when _askPeerAndListen resolves', async () => { + const mockStream = {write: sinon.stub(), end: sinon.stub()}; + const handler = new Handler.ChaincodeMessageHandler(mockStream, mockChaincodeImpl); + const _askPeerAndListenStub = sandbox.stub(handler, '_askPeerAndListen').resolves('some response'); + + const result = await handler.handleWriteBatch([rec], 'theChannelID', 'theTxID'); + + expect(result).to.deep.equal('some response'); + expect(_askPeerAndListenStub.firstCall.args.length).to.deep.equal(2); + expect(_askPeerAndListenStub.firstCall.args[0]).to.deep.equal(expectedMsg); + expect(_askPeerAndListenStub.firstCall.args[1]).to.deep.equal('WriteBatchState'); + }); + + it('should reject when _askPeerAndListen rejects', async () => { + const mockStream = {write: sinon.stub(), end: sinon.stub()}; + const handler = new Handler.ChaincodeMessageHandler(mockStream, mockChaincodeImpl); + const _askPeerAndListenStub = sandbox.stub(handler, '_askPeerAndListen').rejects(); + + const result = handler.handleWriteBatch([rec], 'theChannelID', 'theTxID'); + + await expect(result).to.eventually.be.rejected; + expect(_askPeerAndListenStub.firstCall.args.length).to.deep.equal(2); + expect(_askPeerAndListenStub.firstCall.args[0]).to.deep.equal(expectedMsg); + expect(_askPeerAndListenStub.firstCall.args[1]).to.deep.equal('WriteBatchState'); + }); + }); + + describe('sendBatch', () => { + afterEach(() => { + Handler = rewire('../../../fabric-shim/lib/handler.js'); + sandbox.restore(); + }); + + it('should do nothing for an empty write list', async () => { + const mockStream = {write: sinon.stub(), end: sinon.stub()}; + const handler = new Handler.ChaincodeMessageHandler(mockStream, mockChaincodeImpl); + const handleWriteBatchStub = sandbox.stub(handler, 'handleWriteBatch').resolves(); + + await handler.sendBatch([], 'theChannelID', 'theTxID'); + await handler.sendBatch(null, 'theChannelID', 'theTxID'); + + sinon.assert.notCalled(handleWriteBatchStub); + }); + + it('should send a single batch when under the size limit', async () => { + const mockStream = {write: sinon.stub(), end: sinon.stub()}; + const handler = new Handler.ChaincodeMessageHandler(mockStream, mockChaincodeImpl); + const handleWriteBatchStub = sandbox.stub(handler, 'handleWriteBatch').resolves(); + + await handler.sendBatch(['a'], 'theChannelID', 'theTxID'); + + sinon.assert.calledOnce(handleWriteBatchStub); + expect(handleWriteBatchStub.firstCall.args).to.deep.equal([['a'], 'theChannelID', 'theTxID']); + }); + + it('should split writes that exceed maxSizeWriteBatch', async () => { + const mockStream = {write: sinon.stub(), end: sinon.stub()}; + const handler = new Handler.ChaincodeMessageHandler(mockStream, mockChaincodeImpl); + handler.maxSizeWriteBatch = 2; + const handleWriteBatchStub = sandbox.stub(handler, 'handleWriteBatch').resolves(); + + const writes = ['a', 'b', 'c']; + await handler.sendBatch(writes, 'theChannelID', 'theTxID'); + + sinon.assert.calledTwice(handleWriteBatchStub); + expect(handleWriteBatchStub.firstCall.args).to.deep.equal([['a', 'b'], 'theChannelID', 'theTxID']); + expect(handleWriteBatchStub.secondCall.args).to.deep.equal([['c'], 'theChannelID', 'theTxID']); + }); + }); + describe('handleDeleteState', () => { const key = 'theKey'; const collection = ''; @@ -1739,6 +1896,8 @@ describe('Handler', () => { const createStubStub = sandbox.stub().returns(mockStub); Handler.__set__('createStub', createStubStub); + mockStub.finishWriteBatch.reset(); + mockStub.finishWriteBatch.resolves(); }); afterEach(() => { @@ -1834,6 +1993,7 @@ describe('Handler', () => { expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.notCalled(mockStub.finishWriteBatch); }); it('should handle chaincode.Invoke returning nothing', async () => { @@ -1866,6 +2026,7 @@ describe('Handler', () => { expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.calledOnce(mockStub.finishWriteBatch); }); it ('should handle chaincode.Init returning no status', async () => { @@ -1900,6 +2061,7 @@ describe('Handler', () => { }); expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.notCalled(mockStub.finishWriteBatch); }); it ('should handle chaincode.Invoke returning no status', async () => { @@ -1933,6 +2095,7 @@ describe('Handler', () => { }); expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.calledOnce(mockStub.finishWriteBatch); }); }); @@ -1968,6 +2131,7 @@ describe('Handler', () => { expect(mockHandler.chaincode.Init.firstCall.args[0]).to.deep.equal(mockStub); expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.calledOnce(mockStub.finishWriteBatch); }); it ('should write a COMPLETE message when successful invoke', async () => { @@ -1980,6 +2144,19 @@ describe('Handler', () => { expect(mockHandler.chaincode.Invoke.firstCall.args[0]).to.deep.equal(mockStub); expect(mockHandler._stream.write.calledOnce).to.be.true; expect(mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0])).to.deep.equal(mapFromChaincodeMessage(expectedResponse)); + sinon.assert.calledOnce(mockStub.finishWriteBatch); + }); + + it('should send ERROR when finishWriteBatch fails', async () => { + mockHandler.chaincode.Invoke = sandbox.stub().resolves({status: Stub.RESPONSE_CODE.OK}); + mockStub.finishWriteBatch.rejects(new Error('batch send failed')); + + await handleMessage(msg, mockHandler, 'invoke'); + + expect(mockHandler._stream.write.calledOnce).to.be.true; + const sent = mapFromChaincodeMessage(mockHandler._stream.write.firstCall.args[0]); + expect(sent.type).to.equal(peer.ChaincodeMessage.Type.ERROR); + expect(sent.payload.toString()).to.equal('Error: batch send failed'); }); }); }); diff --git a/libraries/fabric-shim/test/unit/stub.js b/libraries/fabric-shim/test/unit/stub.js index 0ecda526..cc1aedab 100644 --- a/libraries/fabric-shim/test/unit/stub.js +++ b/libraries/fabric-shim/test/unit/stub.js @@ -639,6 +639,15 @@ describe('Stub', () => { expect(handlePutStateStub.calledOnce).to.be.true; expect(handlePutStateStub.firstCall.args).to.deep.equal(['', 'a key', {a:'value'}, 'dummyChannelId', 'dummyTxid']); }); + it('should throw if key is an empty string', async () => { + const handlePutStateStub = sinon.stub().resolves('some state'); + const stub = new Stub({ + handlePutState: handlePutStateStub + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + await expect(stub.putState('', 'a value')).to.be.rejectedWith(/key must not be an empty string/); + sinon.assert.notCalled(handlePutStateStub); + }); }); describe('deleteState', () => { @@ -1149,15 +1158,203 @@ describe('Stub', () => { }); }); - describe('Write Batching Fallbacks', () => { - it('should execute startWriteBatch as a no-op', () => { + describe('Write Batching', () => { + it('should execute startWriteBatch as a no-op when the peer does not support it', () => { const stub = new Stub('dummyClient', 'dummyChannelId', 'dummyTxid', chaincodeInput); stub.startWriteBatch(); + expect(stub.writeBatch).to.equal(null); }); - it('should execute finishWriteBatch as a no-op', async () => { - const stub = new Stub('dummyClient', 'dummyChannelId', 'dummyTxid', chaincodeInput); + it('should call sendBatch when no batch is active', async () => { + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + await stub.finishWriteBatch(); + sinon.assert.calledOnce(sendBatch); + expect(sendBatch.firstCall.args[0]).to.equal(null); + }); + + it('should throw if key is an empty string while batching', async () => { + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({usePeerWriteBatch: true, sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + stub.startWriteBatch(); + await expect(stub.putState('', Buffer.from('value'))).to.be.rejectedWith(/key must not be an empty string/); + await stub.finishWriteBatch(); + expect(sendBatch.firstCall.args[0]).to.deep.equal([]); + }); + + it('should start a write batch when the peer supports it', () => { + const stub = new Stub({usePeerWriteBatch: true}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + stub.startWriteBatch(); + expect(stub.writeBatch).to.not.equal(null); + }); + + it('should not reset an existing write batch', async () => { + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({usePeerWriteBatch: true, sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + stub.startWriteBatch(); + await stub.putState('key1', Buffer.from('value1')); + stub.startWriteBatch(); + await stub.putState('key2', Buffer.from('value2')); + await stub.finishWriteBatch(); + + sinon.assert.calledOnce(sendBatch); + expect(sendBatch.firstCall.args[0]).to.have.length(2); + }); + + it('should queue putState and flush on finishWriteBatch', async () => { + const handlePutState = sinon.stub().resolves('sent'); + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({ + usePeerWriteBatch: true, + handlePutState, + sendBatch + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + stub.startWriteBatch(); + const queued = await stub.putState('a key', 'a value'); + expect(queued).to.equal(undefined); + sinon.assert.notCalled(handlePutState); + + await stub.finishWriteBatch(); + sinon.assert.calledOnce(sendBatch); + expect(sendBatch.firstCall.args[1]).to.equal('dummyChannelId'); + expect(sendBatch.firstCall.args[2]).to.equal('dummyTxid'); + + const records = sendBatch.firstCall.args[0]; + expect(records).to.have.length(1); + expect(records[0].getKey()).to.equal('a key'); + expect(Buffer.from(records[0].getValue_asU8()).toString()).to.equal('a value'); + expect(records[0].getType()).to.equal(peer.WriteRecord.Type.PUT_STATE); + }); + + it('should keep sending immediately when a batch has not been started', async () => { + const handlePutState = sinon.stub().resolves('some state'); + const stub = new Stub({ + usePeerWriteBatch: true, + handlePutState + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + const result = await stub.putState('a key', 'a value'); + expect(result).to.equal('some state'); + sinon.assert.calledOnce(handlePutState); + }); + + it('should overwrite the same data key in the batch', async () => { + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({usePeerWriteBatch: true, sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + stub.startWriteBatch(); + await stub.putState('key1', Buffer.from('first')); + await stub.deleteState('key1'); + await stub.finishWriteBatch(); + + const records = sendBatch.firstCall.args[0]; + expect(records).to.have.length(1); + expect(records[0].getType()).to.equal(peer.WriteRecord.Type.DEL_STATE); + }); + + it('should keep metadata writes independent of data writes', async () => { + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({usePeerWriteBatch: true, sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + const ep = Buffer.from('policy'); + stub.startWriteBatch(); + await stub.putState('key1', Buffer.from('value')); + await stub.setStateValidationParameter('key1', ep); await stub.finishWriteBatch(); + + const records = sendBatch.firstCall.args[0]; + expect(records).to.have.length(2); + const types = records.map((rec) => rec.getType()); + expect(types).to.include(peer.WriteRecord.Type.PUT_STATE); + expect(types).to.include(peer.WriteRecord.Type.PUT_STATE_METADATA); + }); + + it('should queue private data writes and deletes', async () => { + const sendBatch = sinon.stub().resolves(); + const handlePutState = sinon.stub(); + const handleDeleteState = sinon.stub(); + const stub = new Stub({ + usePeerWriteBatch: true, + sendBatch, + handlePutState, + handleDeleteState + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + stub.startWriteBatch(); + await stub.putPrivateData('col', 'key1', 'secret'); + await stub.deletePrivateData('col', 'key2'); + await stub.finishWriteBatch(); + + sinon.assert.notCalled(handlePutState); + sinon.assert.notCalled(handleDeleteState); + const records = sendBatch.firstCall.args[0]; + expect(records).to.have.length(2); + expect(records[0].getCollection()).to.equal('col'); + expect(records[0].getType()).to.equal(peer.WriteRecord.Type.PUT_STATE); + expect(records[1].getType()).to.equal(peer.WriteRecord.Type.DEL_STATE); + }); + + it('should queue purgePrivateData', async () => { + const sendBatch = sinon.stub().resolves(); + const handlePurgeState = sinon.stub(); + const stub = new Stub({ + usePeerWriteBatch: true, + sendBatch, + handlePurgeState + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + stub.startWriteBatch(); + await stub.purgePrivateData('col', 'key1'); + await stub.finishWriteBatch(); + + sinon.assert.notCalled(handlePurgeState); + expect(sendBatch.firstCall.args[0][0].getType()).to.equal(peer.WriteRecord.Type.PURGE_PRIVATE_DATA); + }); + + it('should queue setPrivateDataValidationParameter', async () => { + const sendBatch = sinon.stub().resolves(); + const handlePutStateMetadata = sinon.stub(); + const stub = new Stub({ + usePeerWriteBatch: true, + sendBatch, + handlePutStateMetadata + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + stub.startWriteBatch(); + await stub.setPrivateDataValidationParameter('col', 'key1', Buffer.from('ep')); + await stub.finishWriteBatch(); + + sinon.assert.notCalled(handlePutStateMetadata); + const rec = sendBatch.firstCall.args[0][0]; + expect(rec.getType()).to.equal(peer.WriteRecord.Type.PUT_STATE_METADATA); + expect(rec.getCollection()).to.equal('col'); + }); + + it('should send subsequent writes immediately after finishWriteBatch', async () => { + const handlePutState = sinon.stub().resolves('sent'); + const sendBatch = sinon.stub().resolves(); + const stub = new Stub({ + usePeerWriteBatch: true, + handlePutState, + sendBatch + }, 'dummyChannelId', 'dummyTxid', chaincodeInput); + + stub.startWriteBatch(); + await stub.putState('key1', Buffer.from('value1')); + await stub.finishWriteBatch(); + await stub.putState('key2', Buffer.from('value2')); + + sinon.assert.calledOnce(sendBatch); + sinon.assert.calledOnce(handlePutState); + expect(handlePutState.firstCall.args[1]).to.equal('key2'); + }); + + it('should clear the batch after finishWriteBatch even if sendBatch fails', async () => { + const sendBatch = sinon.stub().rejects(new Error('peer failed')); + const stub = new Stub({usePeerWriteBatch: true, sendBatch}, 'dummyChannelId', 'dummyTxid', chaincodeInput); + stub.startWriteBatch(); + await stub.putState('key1', Buffer.from('value')); + await expect(stub.finishWriteBatch()).to.be.rejectedWith(/peer failed/); + expect(stub.writeBatch).to.equal(null); }); }); @@ -1233,6 +1430,12 @@ describe('Stub', () => { await expect(result).to.eventually.be.rejectedWith(Error, 'key must be a valid string'); }); + it('should throw if key is an empty string', async () => { + const result = stub.putPrivateData('some collection', '', 'some value'); + await expect(result).to.eventually.be.rejectedWith(Error, 'key must not be an empty string'); + sinon.assert.notCalled(handlePutStateStub); + }); + it ('should return handler.handlePutState with string', async () => { const result = await stub.putPrivateData('some collection', 'some key', 'some value'); expect(result).to.deep.equal('some state');