Rework derives to ensure cached registries (#4021)

packages/api-derive/src/chain/getBlock.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { ApiInterfaceRx } from '@polkadot/api/types';
 import type { SignedBlockExtended } from '../type/types';
 
-import { catchError, combineLatest, map, of } from 'rxjs';
+import { catchError, combineLatest, map, of, switchMap } from 'rxjs';
 
 import { createSignedBlockExtended } from '../type';
 import { memo } from '../util';
@@ -25,15 +25,19 @@ import { memo } from '../util';
  */
 export function getBlock (instanceId: string, api: ApiInterfaceRx): (hash: Uint8Array | string) => Observable<SignedBlockExtended | undefined> {
   return memo(instanceId, (hash: Uint8Array | string): Observable<SignedBlockExtended | undefined> =>
-    combineLatest([
+    // we get the block first, setting up the registry
-      api.rpc.chain.getBlock(hash),
+    api.rpc.chain.getBlock(hash).pipe(
-      api.query.system.events.at(hash),
+      switchMap((signedBlock) =>
-      api.query.session
+        combineLatest([
-        ? api.query.session.validators.at(hash)
+          api.query.system.events.at(hash),
-        : of([])
+          api.query.session
-    ]).pipe(
+            ? api.query.session.validators.at(hash)
-      map(([signedBlock, events, validators]): SignedBlockExtended =>
+            : of([])
-        createSignedBlockExtended(api.registry, signedBlock, events, validators)
+        ]).pipe(
+          map(([events, validators]) =>
+            createSignedBlockExtended(api.registry, signedBlock, events, validators)
+          )
+        )
       ),
       catchError((): Observable<undefined> =>
         // where rpc.chain.getHeader throws, we will land here - it can happen that

packages/api-derive/src/chain/getHeader.ts

@@ -3,13 +3,20 @@
 
 import type { Observable } from 'rxjs';
 import type { ApiInterfaceRx } from '@polkadot/api/types';
+import type { AccountId } from '@polkadot/types/interfaces';
 import type { HeaderExtended } from '../type/types';
 
-import { catchError, combineLatest, map, of } from 'rxjs';
+import { catchError, map, of, switchMap } from 'rxjs';
 
 import { createHeaderExtended } from '../type';
 import { memo } from '../util';
 
+function _getValidators (api: ApiInterfaceRx, hash: Uint8Array | string): Observable<AccountId[]> {
+  return api.query.session
+    ? api.query.session.validators.at(hash)
+    : of([] as AccountId[]);
+}
+
 /**
  * @name getHeader
  * @param {( Uint8Array | string )} hash - A block hash as U8 array or string.
@@ -26,14 +33,14 @@ import { memo } from '../util';
  */
 export function getHeader (instanceId: string, api: ApiInterfaceRx): (hash: Uint8Array | string) => Observable<HeaderExtended | undefined> {
   return memo(instanceId, (hash: Uint8Array | string): Observable<HeaderExtended | undefined> =>
-    combineLatest([
+    // we get the header first, setting up the registry
-      api.rpc.chain.getHeader(hash),
+    api.rpc.chain.getHeader(hash).pipe(
-      api.query.session
+      switchMap((header) =>
-        ? api.query.session.validators.at(hash)
+        _getValidators(api, hash).pipe(
-        : of([])
+          map((validators) =>
-    ]).pipe(
+            createHeaderExtended(header.registry, header, validators)
-      map(([header, validators]): HeaderExtended =>
+          )
-        createHeaderExtended(header.registry, header, validators)
+        )
       ),
       catchError((): Observable<undefined> =>
         // where rpc.chain.getHeader throws, we will land here - it can happen that

packages/api-derive/src/chain/subscribeNewBlocks.ts

@@ -3,11 +3,9 @@
 
 import type { Observable } from 'rxjs';
 import type { ApiInterfaceRx } from '@polkadot/api/types';
-import type { Vec } from '@polkadot/types';
+import type { SignedBlockExtended } from '../type/types';
-import type { EventRecord, SignedBlock } from '@polkadot/types/interfaces';
-import type { HeaderExtended, SignedBlockExtended } from '../type/types';
 
-import { combineLatest, map, of, switchMap } from 'rxjs';
+import { map, switchMap } from 'rxjs';
 
 import { createSignedBlockExtended } from '../type';
 import { memo } from '../util';
@@ -19,18 +17,20 @@ import { memo } from '../util';
 export function subscribeNewBlocks (instanceId: string, api: ApiInterfaceRx): () => Observable<SignedBlockExtended> {
   return memo(instanceId, (): Observable<SignedBlockExtended> =>
     api.derive.chain.subscribeNewHeads().pipe(
-      switchMap((header): Observable<[SignedBlock, Vec<EventRecord>, HeaderExtended]> => {
+      switchMap((header) => {
         const blockHash = header.createdAtHash || header.hash;
 
-        return combineLatest(
+        // we get the block first, setting up the registry
-          api.rpc.chain.getBlock(blockHash),
+        return api.rpc.chain.getBlock(blockHash).pipe(
-          api.query.system.events.at(blockHash),
+          switchMap((block) =>
-          of(header)
+            api.query.system.events.at(blockHash).pipe(
+              map((events) =>
+                createSignedBlockExtended(block.registry, block, events, header.validators)
+              )
+            )
+          )
         );
-      }),
+      })
-      map(([block, events, header]) =>
-        createSignedBlockExtended(block.registry, block, events, header.validators)
-      )
     )
   );
 }

packages/api-derive/src/tx/events.ts

@@ -5,7 +5,7 @@ import type { Observable } from 'rxjs';
 import type { ApiInterfaceRx } from '@polkadot/api/types';
 import type { EventRecord, Hash, SignedBlock } from '@polkadot/types/interfaces';
 
-import { combineLatest, map } from 'rxjs';
+import { map, switchMap } from 'rxjs';
 
 import { memo } from '../util';
 
@@ -15,15 +15,17 @@ interface Result {
 }
 
 export function events (instanceId: string, api: ApiInterfaceRx): (at: Hash) => Observable<Result> {
-  return memo(instanceId, (at: Hash): Observable<Result> =>
+  return memo(instanceId, (at: Hash) =>
-    combineLatest([
+    // we get the block first, setting up the registry
-      api.query.system.events.at(at),
+    api.rpc.chain.getBlock(at).pipe(
-      api.rpc.chain.getBlock(at)
+      switchMap((block) =>
-    ]).pipe(
+        api.query.system.events.at(at).pipe(
-      map(([events, block]) => ({
+          map((events): Result => ({
-        block,
+            block,
-        events
+            events
-      }))
+          }))
+        )
+      )
     )
   );
 }