diff --git a/packages/api-derive/src/staking/info.ts b/packages/api-derive/src/staking/info.ts index f69e5cdbe2..9774860bf9 100644 --- a/packages/api-derive/src/staking/info.ts +++ b/packages/api-derive/src/staking/info.ts @@ -83,18 +83,18 @@ function withStashController (api: ApiInterface$Rx, accountId: AccountId, contro const stashId = accountId; return ( - combineLatest([ - eraLength(api)(), - bestNumber(api)(), - api.queryMulti([ - [api.query.session.nextKeyFor, controllerId], - [api.query.staking.ledger, controllerId], - [api.query.staking.nominators, stashId], - [api.query.staking.payee, stashId], - [api.query.staking.stakers, stashId], - [api.query.staking.validators, stashId] - ]) - ]) as any as Observable<[BN, BlockNumber, [Option, Option, [Array], RewardDestination, Exposure, [ValidatorPrefs]]]> + combineLatest([ + eraLength(api)(), + bestNumber(api)(), + api.queryMulti([ + [api.query.session.nextKeyFor, controllerId], + [api.query.staking.ledger, controllerId], + [api.query.staking.nominators, stashId], + [api.query.staking.payee, stashId], + [api.query.staking.stakers, stashId], + [api.query.staking.validators, stashId] + ]) + ]) as any as Observable<[BN, BlockNumber, [Option, Option, [Array], RewardDestination, Exposure, [ValidatorPrefs]]]> ).pipe( map(([eraLength, bestNumber, [nextKeyFor, _stakingLedger, [nominators], rewardDestination, stakers, [validatorPrefs]]]) => { const stakingLedger = _stakingLedger.unwrapOr(null) || undefined; diff --git a/packages/api-derive/src/util/drr.ts b/packages/api-derive/src/util/drr.ts index dbba4b3575..7f03077814 100644 --- a/packages/api-derive/src/util/drr.ts +++ b/packages/api-derive/src/util/drr.ts @@ -2,8 +2,11 @@ // This software may be modified and distributed under the terms // of the Apache-2.0 license. See the LICENSE file for details. -import { distinctUntilChanged, publishReplay, refCount } from 'rxjs/operators'; +import { catchError, distinctUntilChanged, publishReplay, refCount } from 'rxjs/operators'; import { Observable } from 'rxjs'; +import { logger } from '@polkadot/util'; + +const l = logger('drr'); /** * Shorthand for distinctUntilChanged(), publishReplay(1) and refCount(). @@ -12,6 +15,11 @@ import { Observable } from 'rxjs'; */ export const drr = () => (source$: Observable): Observable => source$.pipe( + catchError(err => { + l.error(err); + + throw err; + }), distinctUntilChanged(), publishReplay(1), refCount() diff --git a/packages/api-derive/test/e2e/promise.spec.ts b/packages/api-derive/test/e2e/promise.spec.ts index 13c1556c80..16b48659bf 100644 --- a/packages/api-derive/test/e2e/promise.spec.ts +++ b/packages/api-derive/test/e2e/promise.spec.ts @@ -3,13 +3,15 @@ // of the Apache-2.0 license. See the LICENSE file for details. import ApiPromise from '@polkadot/api/promise/Api'; -import testKeyring from '@polkadot/keyring/testing'; +import testingPairs from '@polkadot/keyring/testingPairs'; import { WsProvider } from '@polkadot/rpc-provider'; +import { RewardDestination } from '@polkadot/types'; import { HeaderExtended } from '../../src/type'; import { DerivedBalances, DerivedFees, DerivedSessionInfo, DerivedStaking } from '../../src/types'; import { SubmittableResult } from '../../../api/src'; +const ALICE_STASH = testingPairs().alice_stash.address; const WS = 'ws://127.0.0.1:9944/'; // const WS = 'wss://poc3-rpc.polkadot.io/'; @@ -121,10 +123,7 @@ describe.skip('derive e2e', () => { describe('verifies derive.staking.unlocking', () => { const UNBOND_VALUE = 1; - const ALICE_STASH = '5GNJqTPyNqANBkUVMN1LPPrxXnFouWXoe2wNSmmEoLctxiZY'; - const ALICE = '5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY'; - const keyring = testKeyring(); - const alicePair = keyring.getPair(ALICE); + const alicePair = testingPairs().alice; it('unbonds dots for Alice (from Alice Stash)', (done) => { return api.tx.staking.unbond(UNBOND_VALUE) @@ -145,10 +144,7 @@ describe.skip('derive e2e', () => { describe('verifies derive.staking.rewardDestination', () => { const PAYEE = 2; - const ALICE_STASH = '5GNJqTPyNqANBkUVMN1LPPrxXnFouWXoe2wNSmmEoLctxiZY'; - const ALICE = '5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY'; - const keyring = testKeyring(); - const alicePair = keyring.getPair(ALICE); + const alicePair = testingPairs().alice; it('Set payee for ALICE to 2', (done) => { return api.tx.staking.setPayee(PAYEE) @@ -171,5 +167,42 @@ describe.skip('derive e2e', () => { done(); }); }); + + it('staking.info updates itself after changing reward destination', async (done) => { + const stashId = testingPairs().alice_stash.address; + + // start by setting the reqred to Staked, so we have a common starting point + await api.tx.staking + .setPayee(new RewardDestination('Staked')) + .signAndSend(testingPairs().alice, ({ status }) => { + if (status.isFinalized) { + console.error('setPayee(Staked) isFinalized'); + } + }); + + let count = 0; // The # of times we got a callback response from api.derive.staking.info + + // Subscribe to staking.info + api.derive.staking.info(stashId, (result) => { + ++count; + + console.error('***', count, JSON.stringify(result)); + + if (count >= 2 && result.rewardDestination!.toString() === 'Stash') { + done(); + } + }).catch(console.error); + + // Wait a bit, and change reward destination + setTimeout(async () => { + await api.tx.staking + .setPayee(new RewardDestination('Stash')) + .signAndSend(testingPairs().alice, ({ status }) => { + if (status.isFinalized) { + console.error('setPayee(Stash) isFinalized'); + } + }); + }, 5000); + }); }); }); diff --git a/packages/api/src/Base.ts b/packages/api/src/Base.ts index 6128150274..f70bcce884 100644 --- a/packages/api/src/Base.ts +++ b/packages/api/src/Base.ts @@ -22,7 +22,7 @@ import { Storage } from '@polkadot/api-metadata/storage/types'; import storageFromMeta from '@polkadot/api-metadata/storage/fromMetadata'; import RpcCore from '@polkadot/rpc-core'; import { WsProvider } from '@polkadot/rpc-provider'; -import { Event, getTypeRegistry, Hash, Metadata, Method, RuntimeVersion, Null, VectorAny } from '@polkadot/types'; +import { Event, getTypeRegistry, Hash, Metadata, Method, RuntimeVersion, Null } from '@polkadot/types'; import Linkage, { LinkageResult } from '@polkadot/types/codec/Linkage'; import { MethodFunction, ModulesWithMethods } from '@polkadot/types/primitive/Method'; import { StorageEntry } from '@polkadot/types/primitive/StorageKey'; @@ -556,9 +556,7 @@ export default abstract class ApiBase { : [arg.creator] as any ); - return this._rpcCore.state - .subscribeStorage(mapped) - .pipe(map((results) => new VectorAny(...results))); + return this._rpcCore.state.subscribeStorage(mapped); }); } @@ -649,7 +647,6 @@ export default abstract class ApiBase { (args: Array) => this._rpcCore.state .subscribeStorage(args.map((arg: CodecArg[] | CodecArg) => [creator, arg])) - .pipe(map((results) => new VectorAny(...results))) ); decorated.size = decorateMethod( diff --git a/packages/api/src/types.ts b/packages/api/src/types.ts index 1100278f45..23baf7bf40 100644 --- a/packages/api/src/types.ts +++ b/packages/api/src/types.ts @@ -95,7 +95,7 @@ export interface StorageEntryObservable { creator: StorageEntry; hash: (arg1?: CodecArg, arg2?: CodecArg) => Observable; key: (arg1?: CodecArg, arg2?: CodecArg) => string; - multi: (args: Array) => Observable; + multi: (args: Array) => Observable>; size: (arg1?: CodecArg, arg2?: CodecArg) => Observable; } @@ -137,7 +137,7 @@ export type QueryableStorageMultiArg = export type QueryableStorageMultiArgs = Array>; export interface QueryableStorageMultiBase { - (calls: QueryableStorageMultiArgs): UnsubscribePromise; + (calls: QueryableStorageMultiArgs): Observable>; } export interface QueryableStorageMultiPromise { diff --git a/packages/rpc-core/src/index.spec.ts b/packages/rpc-core/src/index.spec.ts index 23fc6e8a25..c866bc1de8 100644 --- a/packages/rpc-core/src/index.spec.ts +++ b/packages/rpc-core/src/index.spec.ts @@ -17,7 +17,7 @@ describe('Api', () => { const rpc = new Rpc(new MockProvider()); expect( - Object.keys(rpc) + Object.keys(rpc).filter((key) => !key.startsWith('_')) ).toEqual([ 'provider', 'author', 'chain', 'state', 'system' diff --git a/packages/rpc-core/src/index.ts b/packages/rpc-core/src/index.ts index 785ce76a6c..d308094b50 100644 --- a/packages/rpc-core/src/index.ts +++ b/packages/rpc-core/src/index.ts @@ -11,7 +11,7 @@ import { combineLatest, from, Observable, Observer, of, throwError } from 'rxjs' import { catchError, map, publishReplay, refCount, switchMap } from 'rxjs/operators'; import interfaces from '@polkadot/jsonrpc'; import { Codec } from '@polkadot/types/types'; -import { Option, StorageChangeSet, StorageKey, Vector, createClass, createType } from '@polkadot/types'; +import { Option, StorageChangeSet, StorageData, StorageKey, Vector, createClass, createType } from '@polkadot/types'; import { ExtError, assert, isFunction, isNull, logger } from '@polkadot/util'; const l = logger('rpc-core'); @@ -49,6 +49,7 @@ const EMPTY_META = { * ``` */ export default class Rpc implements RpcInterface { + private _storageCache = new Map>(); readonly provider: ProviderInterface; readonly author: RpcInterface$Section; readonly chain: RpcInterface$Section; @@ -296,20 +297,24 @@ export default class Rpc implements RpcInterface { return createType(type, isNull ? meta.fallback : base, true); } - private formatStorageSet (key: StorageKey, base: StorageChangeSet): Codec | undefined { + private formatStorageSet (key: StorageKey, base: StorageChangeSet): Codec { // Fallback to Data (i.e. just the encoding) if we don't have a specific type const type = key.outputType || 'Data'; - - // see if we have a result value for this specific key const hexKey = key.toHex(); - const { value } = base.changes.find((item) => - item.key.toHex() === hexKey - ) || { value: null }; const meta = key.meta || EMPTY_META; - if (!value) { - return; - } else if (meta.type.isMap && meta.type.asMap.isLinked) { + // see if we have a result value for this specific key, fallback to the cache value + // when the value in the set is not available, or is null/empty. + const { value } = base.changes.find(({ key, value }) => + value.isSome && key.toHex() === hexKey + ) || { value: this._storageCache.get(hexKey) || new Option(StorageData, null) }; + + // store the retrieved result - the only issue with this cache is that there is no + // clearning of it, so very long running processes (not just a couple of hours, longer) + // will increase memory beyond what is allowed. + this._storageCache.set(hexKey, value); + + if (meta.type.isMap && meta.type.asMap.isLinked) { return createType(type, value.unwrapOr(null), true); } else if (meta.modifier.isOptional) { return new Option(