feat: implement native write batching for PutState and related writes

apis/fabric-shim-api/types/index.d.ts

@@ -104,7 +104,10 @@ declare module 'fabric-shim-api' {
         getPrivateDataValidationParameter(collection: string, key: string): Promise<Uint8Array>;
         getPrivateDataByRange(collection: string, startKey: string, endKey: string): Promise<Iterators.StateQueryIterator> & AsyncIterable<Iterators.KV>;
         getPrivateDataByPartialCompositeKey(collection: string, objectType: string, attributes: string[]): Promise<Iterators.StateQueryIterator> & AsyncIterable<Iterators.KV>;
-        getPrivateDataQueryResult(collection: string, query: string): Promise<Iterators.StateQueryIterator> & AsyncIterable<Iterators.KV>;
+        getMultipleStates(keys: string[]): Promise<Uint8Array[]>;
+        getMultiplePrivateData(collection: string, keys: string[]): Promise<Uint8Array[]>;
+        startWriteBatch(): void;
+        finishWriteBatch(): Promise<void>;
     }
 
     interface SplitCompositekey {

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;

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);

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);
     }
 

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 {