protect against stopEvents, fetch past block

This commit is contained in:
Victor Oliva
2024-11-04 12:25:59 +01:00
parent 18370599af
commit 609ea52e98
2 changed files with 121 additions and 19 deletions
+9 -3
View File
@@ -32,7 +32,8 @@ const blockExtrinsics$ = state((hash: string) => {
const decoder$ = dynamicBuilder$.pipe(
map((builder) => createExtrinsicCodec(builder, builder.lookup)),
)
const body$ = blockInfo$(hash).pipe(
const body$ = blockInfoState$(hash).pipe(
filter((v) => !!v),
map((v) => v.body),
filter((v) => !!v),
distinctUntilChanged(),
@@ -181,7 +182,8 @@ export const BlockDetail = () => {
const childBlocks$ = state(
(hash: string) =>
blockInfo$(hash).pipe(
blockInfoState$(hash).pipe(
filter((v) => !!v),
take(1),
switchMap(({ hash, number }) =>
combineKeys(
@@ -195,7 +197,11 @@ const childBlocks$ = state(
: [],
),
),
(hash) => blockInfo$(hash).pipe(startWith({ hash })),
(hash) =>
blockInfoState$(hash).pipe(
filter((v) => !!v),
startWith({ hash }),
),
),
),
map((children) =>
+112 -16
View File
@@ -1,32 +1,43 @@
import { chainHead$ } from "@/chain.state"
import { chainClient$, chainHead$ } from "@/chain.state"
import {
ChainHead$,
PinnedBlocks,
SystemEvent,
} from "@polkadot-api/observable-client"
import { FollowEventWithRuntime } from "@polkadot-api/substrate-client"
import {
FollowEventWithRuntime,
StopError,
} from "@polkadot-api/substrate-client"
import { state, withDefault } from "@react-rxjs/core"
import { partitionByKey } from "@react-rxjs/utils"
import { partitionByKey, toKeySet } from "@react-rxjs/utils"
import { HexString } from "polkadot-api"
import {
catchError,
combineLatest,
concat,
concatMap,
defer,
distinctUntilChanged,
EMPTY,
filter,
forkJoin,
map,
merge,
mergeMap,
NEVER,
Observable,
of,
repeat,
retry,
scan,
skip,
startWith,
Subject,
switchMap,
take,
takeUntil,
takeWhile,
tap,
withLatestFrom,
} from "rxjs"
@@ -106,6 +117,7 @@ export const [blockInfo$, recordedBlocks$] = partitionByKey(
},
),
mergeMap(({ value }) => value),
retryOnStopError(),
),
),
),
@@ -152,22 +164,95 @@ export const [blockInfo$, recordedBlocks$] = partitionByKey(
),
)
export const blockInfoState$ = state((hash: string) => blockInfo$(hash), null)
const getUnpinnedBlockInfo$ = (hash: string): Observable<BlockInfo> => {
const throughRpc$ = chainClient$.pipe(
switchMap((client) =>
defer(() =>
client.client._request<
{
block: {
extrinsics: HexString[]
header: {
digest: { logs: Array<unknown> }
extrinsicsRoot: string
number: HexString
parentHash: HexString
stateRoot: HexString
}
}
} | null,
[string]
>("chain_getBlock", [hash]),
).pipe(
repeat({
delay: 1000,
}),
),
),
filter((v) => !!v),
take(1),
catchError(() => EMPTY),
)
return throughRpc$.pipe(
map(
({ block: { extrinsics, header } }): BlockInfo => ({
hash,
parent: header.parentHash,
body: extrinsics,
events: null,
header: {
digests: header.digest.logs,
extrinsicRoot: header.extrinsicsRoot,
number: Number(header.number),
parentHash: header.parentHash,
stateRoot: header.stateRoot,
},
number: Number(header.number),
status: BlockState.Finalized,
}),
),
tap((v) => disconnectedBlocks$.next(v)),
)
}
export const blockInfoState$ = state(
(hash: string) =>
recordedBlocks$.pipe(
toKeySet(),
map((blocks) => blocks.has(hash)),
distinctUntilChanged(),
switchMap((exists) =>
exists ? blockInfo$(hash) : getUnpinnedBlockInfo$(hash),
),
),
null,
)
const disconnectedBlocks$ = new Subject<BlockInfo>()
export const blocksByHeight$ = state(
recordedBlocks$.pipe(
mergeMap((change) => {
const targets$ = forkJoin(
[...change.keys].map((hash) => blockInfo$(hash).pipe(take(1))),
)
merge(
recordedBlocks$.pipe(
mergeMap((change) => {
const targets$ = forkJoin(
[...change.keys].map((hash) => blockInfo$(hash).pipe(take(1))),
)
return targets$.pipe(
map((targets) => ({
type: change.type,
targets,
})),
)
}),
return targets$.pipe(
map((targets) => ({
type: change.type,
targets,
})),
)
}),
),
disconnectedBlocks$.pipe(
map((block) => ({
type: "add" as const,
targets: [block],
})),
),
).pipe(
scan(
(acc, evt) => {
if (evt.type === "remove") {
@@ -250,6 +335,7 @@ const getBlockStatus$ = (
return of(null)
}),
filter((v) => v !== null),
retryOnStopError(),
takeWhile(
(v) => v !== BlockState.Finalized && v !== BlockState.Pruned,
true,
@@ -258,3 +344,13 @@ const getBlockStatus$ = (
)
}),
)
const retryOnStopError = <T>() =>
retry<T>({
delay(error) {
if (error instanceof StopError) {
return of(null)
}
throw error
},
})