#1547: Fail transfer fulfill with "RESERVED" state and v1.0 content-type

.ncurc.json

@@ -1,5 +1,6 @@
 {
   "reject": [
+    "hapi-swagger",
     "tape",
     "ilp-packet"
   ]

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mojaloop/central-ledger",
-  "version": "11.1.2",
+  "version": "11.1.3",
   "description": "Central ledger hosted by a scheme to record and settle transfers",
   "license": "Apache-2.0",
   "author": "ModusBox",
@@ -107,9 +107,9 @@
     "hapi-swagger": "13.1.0",
     "ilp-packet": "2.2.0",
     "knex": "0.21.4",
-    "lodash": "4.17.19",
+    "lodash": "4.17.20",
     "moment": "2.27.0",
-    "mongoose": "5.9.28",
+    "mongoose": "5.10.0",
     "npm-run-all": "4.1.5",
     "rc": "1.2.8",
     "require-glob": "3.2.0",
@@ -126,7 +126,7 @@
     "jsonpath": "1.0.2",
     "nodemon": "2.0.4",
     "npm-audit-resolver": "2.2.1",
-    "npm-check-updates": "7.1.0",
+    "npm-check-updates": "7.1.1",
     "nyc": "15.1.0",
     "pre-commit": "1.2.2",
     "proxyquire": "2.1.3",

src/handlers/positions/handler.js

@@ -175,14 +175,6 @@ const positions = async (error, messages) => {
         }
         await PositionService.changeParticipantPosition(transferInfo.participantCurrencyId, isReversal, transferInfo.amount, transferStateChange)
         if (action === Enum.Events.Event.Action.RESERVE) {
-          if (message.value.content.headers['content-type'].split('=')[1] === '1.0') {
-            Logger.isInfoEnabled && Logger.info(Utility.breadcrumb(location, `reserve--v1.0--${actionLetter}4.1`))
-            const errorMessage = 'action "RESERVE" is not allowed into position handler for v1.0 clients.'
-            Logger.isErrorEnabled && Logger.error(errorMessage)
-            !!span && span.error(errorMessage)
-            histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
-            return true
-          }
           const transfer = await TransferService.getById(transferInfo.transferId)
           message.value.content.payload = TransferObjectTransform.toFulfil(transfer)
         }

src/handlers/transfers/handler.js

@@ -292,6 +292,18 @@ const fulfil = async (error, messages) => {
     const params = { message, kafkaTopic, decodedPayload: payload, span, consumer: Consumer, producer: Producer }
 
     Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, { path: 'getById' }))
+
+    // We fail early and silently to allow timeout handler abort transfer
+    // if 'RESERVED' transfer state is sent in with v1.0 content-type
+    if (headers['content-type'].split('=')[1] === '1.0' && payload.transferState === TransferState.RESERVED) {
+      Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `failSilentlyforReservedStateWith1.0ContentType--${actionLetter}0`))
+      const errorMessage = 'action "RESERVE" is not allowed in fulfil handler for v1.0 clients.'
+      Logger.isErrorEnabled && Logger.error(errorMessage)
+      !!span && span.error(errorMessage)
+      histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
+      return true
+    }
+
     const transfer = await TransferService.getById(transferId)
     const transferStateEnum = transfer && transfer.transferStateEnumeration
 

test/integration/handlers/handlers.test.js

@@ -143,11 +143,13 @@ const prepareTestData = async (dataObj) => {
 
   const prepareHeaders = {
     'fspiop-source': payer.participant.name,
-    'fspiop-destination': payee.participant.name
+    'fspiop-destination': payee.participant.name,
+    'content-type': 'application/vnd.interoperability.transfers+json;version=1.1'
   }
   const fulfilAbortRejectHeaders = {
     'fspiop-source': payee.participant.name,
-    'fspiop-destination': payer.participant.name
+    'fspiop-destination': payer.participant.name,
+    'content-type': 'application/vnd.interoperability.transfers+json;version=1.1'
   }
 
   const fulfilPayload = {

test/unit/handlers/positions/handler.test.js

@@ -304,32 +304,6 @@ Test('Position handler', transferHandlerTest => {
       }
     })
 
-    positionsTest.test('log error when RESERVED transfer state is received from v1.0 clients', async (test) => {
-      const isIncrease = false
-      const transferStateChange = {
-        transferId: transferInfo.transferId,
-        transferStateId: TransferState.RESERVED
-      }
-
-      await Consumer.createHandler(topicName, config, command)
-      Kafka.transformGeneralTopicName.returns(topicName)
-      Kafka.getKafkaConfig.returns(config)
-
-      const m = Object.assign({}, MainUtil.clone(messages[1]))
-      TransferService.getTransferInfoToChangePosition.withArgs(m.value.content.uriParams.id, Enum.Accounts.TransferParticipantRoleType.PAYEE_DFSP, Enum.Accounts.LedgerEntryType.PRINCIPLE_VALUE).returns(transferInfo)
-      TransferStateChange.saveTransferStateChange.resolves(true)
-      PositionService.changeParticipantPosition.withArgs(transferInfo.participantCurrencyId, isIncrease, transferInfo.amount, transferStateChange).resolves(true)
-      m.value.metadata.event.action = transferEventAction.RESERVE
-      m.value.content.headers = { 'content-type': 'application/vnd.interoperability.transfers+json;version=1.0' }
-      Kafka.proceed.returns(true)
-
-      const result = await allTransferHandlers.positions(null, m)
-      Logger.info(result)
-      test.ok(SpanStub.error.calledWith('action "RESERVE" is not allowed into position handler for v1.0 clients.'))
-      test.equal(result, true)
-      test.end()
-    })
-
     positionsTest.end()
   })