storage-node-v2: Update runtime api usage.

- protect nonce using async lock
- use single api initialization point

storage-node-v2/package.json

@@ -15,6 +15,7 @@
     "@polkadot/api": "4.2.1",
     "@types/express": "4.17.1",
     "@types/multer": "^1.4.5",
+    "await-lock": "^2.1.0",
     "blake3": "^2.1.4",
     "express": "4.17.1",
     "express-openapi-validator": "^4.12.4",

storage-node-v2/scripts/create-several-buckets.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+
+yarn storage-node dev:init
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+curl http://localhost:3333/test &
+
+
+# yarn storage-node leader:create-bucket -i=0 -a --dev &
+# yarn storage-node leader:create-bucket -i=0 -a --dev &
+# yarn storage-node leader:create-bucket -i=0 -a --dev &
+# yarn storage-node leader:create-bucket -i=0 -a --dev &
+# yarn storage-node leader:create-bucket -i=0 -a --dev

storage-node-v2/src/command-base/ApiCommandBase.ts

@@ -2,8 +2,11 @@ import { Command, flags } from '@oclif/command'
 import { createApi, getAlicePair } from '../services/runtime/api'
 import { getAccountFromJsonFile } from '../services/runtime/accounts'
 import { KeyringPair } from '@polkadot/keyring/types'
+import { ApiPromise } from '@polkadot/api'
 
 export default abstract class ApiCommandBase extends Command {
+  private api: ApiPromise | null = null
+
   static keyflags = {
     help: flags.help({ char: 'h' }),
     dev: flags.boolean({ char: 'd', description: 'Use development mode' }),
@@ -25,8 +28,22 @@ export default abstract class ApiCommandBase extends Command {
     super.finally(err)
   }
 
+  protected async getApi(): Promise<ApiPromise> {
+    if (this.api === null) {
+      throw new Error('Runtime Api is uninitialized.')
+    }
+
+    return this.api
+  }
+
+  async init(): Promise<void> {
+    this.api = await createApi()
+
+    await this.getApi()
+  }
+
   async ensureDevelopmentChain(): Promise<void> {
-    const api = await createApi()
+    const api = await this.getApi()
 
     const developmentChainName = 'Development'
     const runningChainName = await api.rpc.system.chain()

storage-node-v2/src/commands/dev/init.ts

@@ -2,7 +2,7 @@ import { flags } from '@oclif/command'
 import { hireStorageWorkingGroupLead } from '../../services/runtime/hireLead'
 import ApiCommandBase from '../../command-base/ApiCommandBase'
 
-//TODO: consider renaming the command - DevHireLeader
+// TODO: consider renaming the command - DevHireLeader
 export default class DevInit extends ApiCommandBase {
   static description =
     'Initialize development environment. Sets Alice as storage working group leader.'
@@ -14,6 +14,7 @@ export default class DevInit extends ApiCommandBase {
   async run(): Promise<void> {
     await this.ensureDevelopmentChain()
 
-    await hireStorageWorkingGroupLead()
+    const api = await this.getApi()
+    await hireStorageWorkingGroupLead(api)
   }
 }

storage-node-v2/src/commands/dev/upload.ts

@@ -29,6 +29,7 @@ export default class DevUpload extends ApiCommandBase {
 
     this.log('Uploading data objects...')
 
-    await uploadDataObjects(objectSize, objectCid)
+    const api = await this.getApi()
+    await uploadDataObjects(api, objectSize, objectCid)
   }
 }

storage-node-v2/src/commands/leader/create-bucket.ts

@@ -36,8 +36,10 @@ export default class LeaderCreateBucket extends ApiCommandBase {
     }
 
     const account = this.getAccount(flags)
+    const api = await this.getApi()
 
     await createStorageBucket(
+      api,
       account,
       invitedWorker,
       allowNewBags,

storage-node-v2/src/commands/leader/update-bag-limit.ts

@@ -26,6 +26,7 @@ export default class LeaderUpdateBagLimit extends ApiCommandBase {
     const account = this.getAccount(flags)
     const limit = flags.limit ?? 0
 
-    await updateStorageBucketsPerBagLimit(account, limit)
+    const api = await this.getApi()
+    await updateStorageBucketsPerBagLimit(api, account, limit)
   }
 }

storage-node-v2/src/commands/leader/update-bag.ts

@@ -49,6 +49,7 @@ export default class LeaderUpdateBag extends ApiCommandBase {
     const account = this.getAccount(flags)
 
     // TODO: add bag parameter
-    await updateStorageBucketsForBag(account, bucket, flags.remove)
+    const api = await this.getApi()
+    await updateStorageBucketsForBag(api, account, bucket, flags.remove)
   }
 }

storage-node-v2/src/commands/leader/update-voucher-limits.ts

@@ -32,6 +32,12 @@ export default class LeaderUpdateVoucherLimits extends ApiCommandBase {
     const objectsLimit = flags.objects ?? 0
     const sizeLimit = flags.size ?? 0
 
-    await updateStorageBucketsVoucherMaxLimits(account, sizeLimit, objectsLimit)
+    const api = await this.getApi()
+    await updateStorageBucketsVoucherMaxLimits(
+      api,
+      account,
+      sizeLimit,
+      objectsLimit
+    )
   }
 }

storage-node-v2/src/commands/operator/accept-invitation.ts

@@ -32,6 +32,7 @@ export default class OperatorAcceptInvitation extends ApiCommandBase {
 
     const account = this.getAccount(flags)
 
-    await acceptStorageBucketInvitation(account, worker, bucket)
+    const api = await this.getApi()
+    await acceptStorageBucketInvitation(api, account, worker, bucket)
   }
 }

storage-node-v2/src/commands/server.ts

@@ -32,10 +32,11 @@ export default class Server extends ApiCommandBase {
     }
 
     const account = this.getAccount(flags)
+    const api = await this.getApi()
 
     try {
       const port = flags.port
-      const app = await createApp(account, flags.uploads)
+      const app = await createApp(api, account, flags.uploads)
       console.info(`Listening on http://localhost:${port}`)
       app.listen(port)
     } catch (err) {

storage-node-v2/src/services/runtime/api.ts

@@ -11,7 +11,7 @@ import {
   DispatchResult,
 } from '@polkadot/types/interfaces/system'
 import { Keyring } from '@polkadot/keyring'
-
+import { getNonce } from './nonceKeeper'
 // TODO: ApiHelper class container for functions ???
 
 export class ExtrinsicFailedError extends Error {}
@@ -107,7 +107,7 @@ export async function sendAndFollowTx(
 ): Promise<boolean> {
   try {
     // TODO: use async-lock package
-    const nonce = await api.rpc.system.accountNextIndex(account.address)
+    const nonce = await getNonce(api, account)
 
     await sendExtrinsic(api, account, tx, nonce)
     console.log(chalk.green(`Extrinsic successful!`))

storage-node-v2/src/services/runtime/extrinsics.ts

@@ -1,13 +1,14 @@
 import {
-  createApi,
   sendAndFollowSudoNamedTx,
   sendAndFollowNamedTx,
   getAlicePair,
 } from './api'
 import { KeyringPair } from '@polkadot/keyring/types'
 import { CodecArg } from '@polkadot/types/types'
+import { ApiPromise } from '@polkadot/api'
 
 export async function createStorageBucket(
+  api: ApiPromise,
   account: KeyringPair,
   invitedWorker: number | null = null,
   allowedNewBags = true,
@@ -15,8 +16,6 @@ export async function createStorageBucket(
   objectsLimit = 0
 ): Promise<void> {
   try {
-    const api = await createApi()
-
     const invitedWorkerValue = api.createType('Option<WorkerId>', invitedWorker)
 
     await sendAndFollowNamedTx(api, account, 'storage', 'createStorageBucket', [
@@ -31,13 +30,12 @@ export async function createStorageBucket(
 }
 
 export async function acceptStorageBucketInvitation(
+  api: ApiPromise,
   account: KeyringPair,
   workerId: number,
   storageBucketId: number
 ): Promise<void> {
   try {
-    const api = await createApi()
-
     await sendAndFollowNamedTx(
       api,
       account,
@@ -53,13 +51,12 @@ export async function acceptStorageBucketInvitation(
 // TODO: Add dynamic bag parameter
 
 export async function updateStorageBucketsForBag(
+  api: ApiPromise,
   account: KeyringPair,
   bucketId: number,
   removeBucket: boolean
 ): Promise<void> {
   try {
-    const api = await createApi()
-
     let addBuckets: CodecArg
     let removeBuckets: CodecArg
 
@@ -82,12 +79,11 @@ export async function updateStorageBucketsForBag(
 }
 
 export async function uploadDataObjects(
+  api: ApiPromise,
   objectSize: number,
   objectCid: string
 ): Promise<void> {
   try {
-    const api = await createApi()
-
     const alice = getAlicePair()
 
     const data = api.createType('UploadParameters', {
@@ -113,14 +109,13 @@ export async function uploadDataObjects(
 }
 
 export async function acceptPendingDataObjects(
+  api: ApiPromise,
   account: KeyringPair,
   workerId: number,
   storageBucketId: number,
   dataObjects: number[]
 ): Promise<void> {
   try {
-    const api = await createApi()
-
     const bagId = { 'Static': 'Council' }
 
     const dataObjectSet: CodecArg = api.createType(
@@ -142,11 +137,11 @@ export async function acceptPendingDataObjects(
 }
 
 export async function updateStorageBucketsPerBagLimit(
+  api: ApiPromise,
   account: KeyringPair,
   newLimit: number
 ): Promise<void> {
   try {
-    const api = await createApi()
     await sendAndFollowNamedTx(
       api,
       account,
@@ -160,12 +155,12 @@ export async function updateStorageBucketsPerBagLimit(
 }
 
 export async function updateStorageBucketsVoucherMaxLimits(
+  api: ApiPromise,
   account: KeyringPair,
   newSizeLimit: number,
   newObjectLimit: number
 ): Promise<void> {
   try {
-    const api = await createApi()
     await sendAndFollowNamedTx(
       api,
       account,

storage-node-v2/src/services/runtime/hireLead.ts

@@ -1,5 +1,4 @@
 import {
-  createApi,
   getAlicePair,
   sendAndFollowSudoNamedTx,
   sendAndFollowNamedTx,
@@ -12,23 +11,24 @@ import {
   ApplicationId,
 } from '@joystream/types/working-group'
 import { MemberId } from '@joystream/types/members'
+import { ApiPromise } from '@polkadot/api'
 
-export async function hireStorageWorkingGroupLead(): Promise<void> {
-  const api = await createApi()
-
+export async function hireStorageWorkingGroupLead(
+  api: ApiPromise
+): Promise<void> {
   const SudoKeyPair = getAlicePair()
   const LeadKeyPair = getAlicePair()
 
-  const nullValue = (null as unknown) as CodecArg
+  const nullValue = null as unknown as CodecArg
 
   // Create membership if not already created
   const members = (await api.query.members.memberIdsByControllerAccountId(
     LeadKeyPair.address
   )) as Vec<MemberId>
 
-  let memberId:
-    | bigint
-    | undefined = (members.toArray()[0] as MemberId)?.toBigInt()
+  let memberId: bigint | undefined = (
+    members.toArray()[0] as MemberId
+  )?.toBigInt()
 
   if (memberId === undefined) {
     console.log('Preparing member account creation extrinsic...')
@@ -42,9 +42,8 @@ export async function hireStorageWorkingGroupLead(): Promise<void> {
   }
 
   // Create a new lead opening.
-  const currentLead = (await api.query.storageWorkingGroup.currentLead()) as Option<
-    WorkerId
-  >
+  const currentLead =
+    (await api.query.storageWorkingGroup.currentLead()) as Option<WorkerId>
   if (currentLead.isSome) {
     console.log('Storage lead already exists, skipping...')
     return
@@ -52,8 +51,12 @@ export async function hireStorageWorkingGroupLead(): Promise<void> {
 
   console.log(`Making member id: ${memberId} the content lead.`)
 
-  const newOpeningId = ((await api.query.storageWorkingGroup.nextOpeningId()) as OpeningId).toBigInt()
-  const newApplicationId = ((await api.query.storageWorkingGroup.nextApplicationId()) as ApplicationId).toBigInt()
+  const newOpeningId = (
+    (await api.query.storageWorkingGroup.nextOpeningId()) as OpeningId
+  ).toBigInt()
+  const newApplicationId = (
+    (await api.query.storageWorkingGroup.nextApplicationId()) as ApplicationId
+  ).toBigInt()
 
   // Create curator lead opening
   console.log('Preparing Create Storage Lead Opening extrinsic...')

storage-node-v2/src/services/runtime/nonceKeeper.ts

@@ -0,0 +1,28 @@
+import { KeyringPair } from '@polkadot/keyring/types'
+import type { Index } from '@polkadot/types/interfaces/runtime'
+import BN from 'bn.js'
+import AwaitLock from 'await-lock'
+import { ApiPromise } from '@polkadot/api'
+
+let nonce: Index | null = null
+const lock = new AwaitLock()
+
+export async function getNonce(
+  api: ApiPromise,
+  account: KeyringPair
+): Promise<Index> {
+  await lock.acquireAsync()
+  try {
+    if (nonce === null) {
+      nonce = await api.rpc.system.accountNextIndex(account.address)
+    } else {
+      nonce = nonce.add(new BN(1)) as Index
+    }
+  } finally {
+    lock.release()
+  }
+
+  console.log(`Last nonce:${nonce}`)
+
+  return nonce as Index
+}

storage-node-v2/src/services/webApi/app.ts

@@ -5,12 +5,15 @@ import { Express, NextFunction } from 'express-serve-static-core'
 import * as OpenApiValidator from 'express-openapi-validator'
 import { OpenAPIV3 } from 'express-openapi-validator/dist/framework/types'
 import { KeyringPair } from '@polkadot/keyring/types'
+import { ApiPromise } from '@polkadot/api'
 import { TokenRequest, verifyTokenSignature } from '../auth'
+import { createStorageBucket } from '../runtime/extrinsics'
 
 // TODO: custom errors (including validation errors)
 // TODO: custom authorization errors
 
 export async function createApp(
+  api: ApiPromise,
   account: KeyringPair,
   uploadsDir: string
 ): Promise<Express> {
@@ -24,10 +27,17 @@ export async function createApp(
   // TODO: check path
   app.use('/files', express.static(uploadsDir))
 
+  app.get('/test', async function (req, res) {
+    await createStorageBucket(api, account)
+
+    res.send('ok')
+  })
+
   app.use(
     // Set parameters for each request.
     (req: express.Request, res: express.Response, next: NextFunction) => {
       res.locals.storageProviderAccount = account
+      res.locals.api = api
       next()
     },
     OpenApiValidator.middleware({

storage-node-v2/src/services/webApi/controllers/publicApi.ts

@@ -3,6 +3,7 @@ import { acceptPendingDataObjects } from '../../runtime/extrinsics'
 import { TokenRequest, signToken } from '../../auth'
 import { hashFile } from '../../../services/hashing'
 import { KeyringPair } from '@polkadot/keyring/types'
+import { ApiPromise } from '@polkadot/api'
 import fs from 'fs'
 const fsPromises = fs.promises
 
@@ -34,6 +35,7 @@ export async function upload(
     await fsPromises.rename(fileObj.path, newPath)
 
     await acceptPendingDataObjects(
+      getApi(res),
       getAccount(res),
       uploadRequest.workerId,
       uploadRequest.storageBucketId,
@@ -83,6 +85,14 @@ function getAccount(res: express.Response): KeyringPair {
   throw new Error('No Joystream account loaded.')
 }
 
+function getApi(res: express.Response): ApiPromise {
+  if (res.locals.api) {
+    return res.locals.api
+  }
+
+  throw new Error('No Joystream API loaded.')
+}
+
 function getTokenRequest(req: express.Request): TokenRequest {
   const tokenRequest = req.body as TokenRequest
   if (tokenRequest) {

yarn.lock

@@ -7513,6 +7513,11 @@ autoprefixer@^9.5.1, autoprefixer@^9.7.2, autoprefixer@^9.8.0:
     postcss "^7.0.32"
     postcss-value-parser "^4.1.0"
 
+await-lock@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/await-lock/-/await-lock-2.1.0.tgz#bc78c51d229a34d5d90965a1c94770e772c6145e"
+  integrity sha512-t7Zm5YGgEEc/3eYAicF32m/TNvL+XOeYZy9CvBUeJY/szM7frLolFylhrlZNWV/ohWhcUXygrBGjYmoQdxF4CQ==
+
 aws-credstash@^3.0.0:
   version "3.0.0"
   resolved "https://registry.yarnpkg.com/aws-credstash/-/aws-credstash-3.0.0.tgz#377de983c149a8a5471e1ff23c4550d35514b726"