refactor: use top-level client to get blocks
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
import { CopyText } from "@/components/Copy"
|
||||
import { Popover } from "@/components/Popover"
|
||||
import { Link } from "@/hashParams"
|
||||
import { chainHead$ } from "@/state/chains/chain.state"
|
||||
import { client$ } from "@/state/chains/chain.state"
|
||||
import { state, useStateObservable } from "@react-rxjs/core"
|
||||
import { FC } from "react"
|
||||
import { combineLatest, debounceTime, map, switchMap } from "rxjs"
|
||||
@@ -10,7 +10,9 @@ import { BlockInfo, blocksByHeight$, finalized$ } from "./block.state"
|
||||
import { BlockPopover } from "./BlockPopover"
|
||||
import * as Finalizing from "./FinalizingTable"
|
||||
|
||||
const best$ = chainHead$.pipeState(switchMap((chainHead) => chainHead.best$))
|
||||
const best$ = client$.pipeState(
|
||||
switchMap((client) => client.bestBlocks$.pipe(map(([best]) => best))),
|
||||
)
|
||||
|
||||
interface PositionedBlock {
|
||||
block: BlockInfo
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { chainHead$ } from "@/state/chains/chain.state"
|
||||
import { CircularProgress } from "@/components/CircularProgress"
|
||||
import { client$ } from "@/state/chains/chain.state"
|
||||
import { state, useStateObservable } from "@react-rxjs/core"
|
||||
import {
|
||||
animationFrames,
|
||||
@@ -10,7 +10,9 @@ import {
|
||||
} from "rxjs"
|
||||
import { targetBlockTime$ } from "./blockTime.state"
|
||||
|
||||
const best$ = chainHead$.pipeState(switchMap((chainHead) => chainHead.best$))
|
||||
const best$ = client$.pipeState(
|
||||
switchMap((client) => client.bestBlocks$.pipe(map(([best]) => best))),
|
||||
)
|
||||
const bestBlockTime$ = best$.pipeState(
|
||||
// Only count when increasing height
|
||||
map((block) => block.number),
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { CircularProgress } from "@/components/CircularProgress"
|
||||
import { groupBy } from "@/lib/groupBy"
|
||||
import { chainClient$, client$, runtimeCtx$ } from "@/state/chains/chain.state"
|
||||
import { state, useStateObservable } from "@react-rxjs/core"
|
||||
import {
|
||||
catchError,
|
||||
@@ -11,17 +13,11 @@ import {
|
||||
withLatestFrom,
|
||||
} from "rxjs"
|
||||
import { targetBlockTime$ } from "./blockTime.state"
|
||||
import {
|
||||
chainClient$,
|
||||
chainHead$,
|
||||
runtimeCtx$,
|
||||
} from "@/state/chains/chain.state"
|
||||
import { groupBy } from "@/lib/groupBy"
|
||||
|
||||
const bestBlock$ = chainHead$.pipeState(
|
||||
const bestBlock$ = client$.pipeState(
|
||||
// Only count when increasing height
|
||||
switchMap((chainHead) => chainHead.best$),
|
||||
map((block) => block.number),
|
||||
switchMap((client) => client.bestBlocks$),
|
||||
map(([block]) => block.number),
|
||||
distinctUntilChanged((prev, current) => prev >= current),
|
||||
)
|
||||
|
||||
|
||||
@@ -1,18 +1,18 @@
|
||||
import { chainHead$, runtimeCtx$ } from "@/state/chains/chain.state"
|
||||
import { client$, runtimeCtx$ } from "@/state/chains/chain.state"
|
||||
import { useStateObservable, withDefault } from "@react-rxjs/core"
|
||||
import { FC, PropsWithChildren } from "react"
|
||||
import { map, switchMap } from "rxjs"
|
||||
import { twMerge } from "tailwind-merge"
|
||||
import { BlockTime } from "./BlockTime"
|
||||
import { EpochRemainingTime } from "./EpochTime"
|
||||
import { useStateObservable, withDefault } from "@react-rxjs/core"
|
||||
|
||||
const finalized$ = chainHead$.pipeState(
|
||||
switchMap((chainHead) => chainHead.finalized$),
|
||||
const finalized$ = client$.pipeState(
|
||||
switchMap((chainHead) => chainHead.finalizedBlock$),
|
||||
map((v) => v.number.toLocaleString()),
|
||||
)
|
||||
const best$ = chainHead$.pipeState(
|
||||
switchMap((chainHead) => chainHead.best$),
|
||||
map((v) => v.number.toLocaleString()),
|
||||
const best$ = client$.pipeState(
|
||||
switchMap((chainHead) => chainHead.bestBlocks$),
|
||||
map(([v]) => v.number.toLocaleString()),
|
||||
)
|
||||
|
||||
// epoch is only available for relay chains
|
||||
|
||||
@@ -1,32 +1,25 @@
|
||||
import { chainClient$, chainHead$ } from "@/state/chains/chain.state"
|
||||
import {
|
||||
ChainHead$,
|
||||
PinnedBlocks,
|
||||
SystemEvent,
|
||||
} from "@polkadot-api/observable-client"
|
||||
import { StopError } from "@polkadot-api/substrate-client"
|
||||
import { state, withDefault } from "@react-rxjs/core"
|
||||
import { chainClient$, client$ } from "@/state/chains/chain.state"
|
||||
import { SystemEvent } from "@polkadot-api/observable-client"
|
||||
import { state } from "@react-rxjs/core"
|
||||
import { partitionByKey, toKeySet } from "@react-rxjs/utils"
|
||||
import { HexString } from "polkadot-api"
|
||||
import { HexString, PolkadotClient } from "polkadot-api"
|
||||
import {
|
||||
catchError,
|
||||
combineLatest,
|
||||
concat,
|
||||
concatMap,
|
||||
defer,
|
||||
distinctUntilChanged,
|
||||
EMPTY,
|
||||
filter,
|
||||
forkJoin,
|
||||
from,
|
||||
map,
|
||||
merge,
|
||||
mergeMap,
|
||||
NEVER,
|
||||
Observable,
|
||||
ObservedValueOf,
|
||||
of,
|
||||
repeat,
|
||||
retry,
|
||||
scan,
|
||||
skip,
|
||||
startWith,
|
||||
@@ -40,9 +33,8 @@ import {
|
||||
withLatestFrom,
|
||||
} from "rxjs"
|
||||
|
||||
export const finalized$ = chainHead$.pipeState(
|
||||
switchMap((chainHead) => chainHead.finalized$),
|
||||
withDefault(null),
|
||||
export const finalized$ = client$.pipeState(
|
||||
switchMap((client) => client.finalizedBlock$),
|
||||
)
|
||||
|
||||
export enum BlockState {
|
||||
@@ -67,101 +59,53 @@ export interface BlockInfo {
|
||||
status: BlockState
|
||||
}
|
||||
export const [blockInfo$, recordedBlocks$] = partitionByKey(
|
||||
chainHead$.pipe(
|
||||
switchMap((chainHead) =>
|
||||
chainHead.follow$.pipe(
|
||||
withInitializedNumber(),
|
||||
scan(
|
||||
(acc, evt) => {
|
||||
switch (evt.type) {
|
||||
case "initialized": {
|
||||
const blockNumbers: Record<string, number> = {}
|
||||
evt.finalizedBlockHashes.forEach((hash, i) => {
|
||||
const parent = blockNumbers[evt.finalizedBlockHashes[i - 1]]
|
||||
blockNumbers[hash] = parent != null ? parent + 1 : evt.number
|
||||
})
|
||||
return {
|
||||
value: evt.finalizedBlockHashes.map((hash, i) => ({
|
||||
hash,
|
||||
parent: evt.finalizedBlockHashes[i - 1] || evt.parentHash,
|
||||
number: blockNumbers[hash],
|
||||
})),
|
||||
blockNumbers,
|
||||
}
|
||||
}
|
||||
case "newBlock": {
|
||||
const number = acc.blockNumbers[evt.parentBlockHash] + 1
|
||||
acc.blockNumbers[evt.blockHash] = number
|
||||
return {
|
||||
value: [
|
||||
{
|
||||
hash: evt.blockHash,
|
||||
parent: evt.parentBlockHash,
|
||||
number,
|
||||
},
|
||||
],
|
||||
blockNumbers: acc.blockNumbers,
|
||||
}
|
||||
}
|
||||
}
|
||||
return { value: [], blockNumbers: acc.blockNumbers }
|
||||
},
|
||||
{ value: [], blockNumbers: {} } as {
|
||||
value: Array<{
|
||||
hash: string
|
||||
parent: string
|
||||
number: number
|
||||
}>
|
||||
blockNumbers: Record<string, number>
|
||||
},
|
||||
),
|
||||
mergeMap(({ value }) => value),
|
||||
retryOnStopError(),
|
||||
),
|
||||
),
|
||||
),
|
||||
client$.pipe(switchMap((client) => client.blocks$)),
|
||||
(v) => v.hash,
|
||||
(initialized$) =>
|
||||
initialized$.pipe(
|
||||
(block$) =>
|
||||
block$.pipe(
|
||||
take(1),
|
||||
withLatestFrom(chainHead$),
|
||||
withLatestFrom(client$),
|
||||
switchMap(
|
||||
([{ hash, parent, number }, chainHead]): Observable<BlockInfo> =>
|
||||
([{ hash, parent, number }, client]): Observable<BlockInfo> =>
|
||||
concat(
|
||||
combineLatest({
|
||||
hash: of(hash),
|
||||
parent: of(parent),
|
||||
number: of(number),
|
||||
body: chainHead.body$(hash).pipe(
|
||||
body: from(client.getBlockBody(hash)).pipe(
|
||||
startWith(null),
|
||||
catchError((err) => {
|
||||
console.error("fetch body failed", err)
|
||||
return of(null)
|
||||
}),
|
||||
),
|
||||
events: chainHead.eventsAt$(hash).pipe(
|
||||
events: from(
|
||||
client.getUnsafeApi().query.System.Events.getValue({
|
||||
at: hash,
|
||||
}),
|
||||
).pipe(
|
||||
startWith(null),
|
||||
catchError((err) => {
|
||||
console.error("fetch events failed", err)
|
||||
return of(null)
|
||||
}),
|
||||
),
|
||||
header: chainHead.header$(hash).pipe(
|
||||
header: from(client.getBlockHeader(hash)).pipe(
|
||||
startWith(null),
|
||||
catchError((err) => {
|
||||
console.error("fetch header failed", err)
|
||||
return of(null)
|
||||
}),
|
||||
),
|
||||
status: getBlockStatus$(chainHead, hash, number),
|
||||
status: getBlockStatus$(client, hash, number),
|
||||
}),
|
||||
NEVER,
|
||||
),
|
||||
),
|
||||
takeUntil(
|
||||
merge(
|
||||
// Reset when chainHead is changed
|
||||
chainHead$.pipe(skip(1)),
|
||||
// Reset when client is changed
|
||||
client$.pipe(skip(1)),
|
||||
// Or after 1 hour
|
||||
timer(60 * 60 * 1000),
|
||||
),
|
||||
@@ -295,81 +239,29 @@ export const blocksByHeight$ = state(
|
||||
),
|
||||
)
|
||||
|
||||
function withInitializedNumber() {
|
||||
return (source$: Observable<ObservedValueOf<ChainHead$["follow$"]>>) =>
|
||||
source$.pipe(
|
||||
withLatestFrom(chainHead$),
|
||||
concatMap(([event, chainHead]) => {
|
||||
return event.type !== "initialized"
|
||||
? of(event)
|
||||
: chainHead.header$(event.finalizedBlockHashes[0]).pipe(
|
||||
map((header) => ({
|
||||
...event,
|
||||
number: header.number,
|
||||
parentHash: header.parentHash,
|
||||
})),
|
||||
)
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
const getBlockStatus$ = (
|
||||
chainHead: ChainHead$,
|
||||
client: PolkadotClient,
|
||||
hash: string,
|
||||
number: number,
|
||||
): Observable<BlockState> =>
|
||||
chainHead.pinnedBlocks$.pipe(
|
||||
take(1),
|
||||
switchMap((pinnedBlocks) => {
|
||||
const block = (hash: string) => pinnedBlocks.blocks.get(hash)
|
||||
const blockNum = (hash: string) => block(hash)?.number
|
||||
const finalized = blockNum(pinnedBlocks.finalized)!
|
||||
if (number <= finalized) {
|
||||
// assume we are in the list of finalized blocks on `initialized`
|
||||
return of(BlockState.Finalized)
|
||||
}
|
||||
|
||||
const getInitialState = (pinnedBlocks: PinnedBlocks) => {
|
||||
let bestBranch = pinnedBlocks.best
|
||||
while (blockNum(bestBranch)! > finalized && hash !== bestBranch) {
|
||||
bestBranch = block(bestBranch)!.parent
|
||||
}
|
||||
return block(bestBranch)?.hash === hash
|
||||
? BlockState.Best
|
||||
: BlockState.Fork
|
||||
}
|
||||
const initialState = getInitialState(pinnedBlocks)
|
||||
|
||||
return chainHead.follow$.pipe(
|
||||
concatMap((evt) => {
|
||||
switch (evt.type) {
|
||||
case "bestBlockChanged":
|
||||
return chainHead.pinnedBlocks$.pipe(take(1), map(getInitialState))
|
||||
case "finalized":
|
||||
if (evt.finalizedBlockHashes.includes(hash))
|
||||
return of(BlockState.Finalized)
|
||||
if (evt.prunedBlockHashes.includes(hash))
|
||||
return of(BlockState.Pruned)
|
||||
}
|
||||
return of(null)
|
||||
}),
|
||||
filter((v) => v !== null),
|
||||
retryOnStopError(),
|
||||
takeWhile(
|
||||
(v) => v !== BlockState.Finalized && v !== BlockState.Pruned,
|
||||
true,
|
||||
),
|
||||
startWith(initialState),
|
||||
)
|
||||
}),
|
||||
merge(
|
||||
client.finalizedBlock$.pipe(
|
||||
take(1),
|
||||
// If the latest finalized is ahead, assume it is finalized (?)
|
||||
filter((b) => b.number > number),
|
||||
map(() => BlockState.Finalized),
|
||||
),
|
||||
combineLatest([client.bestBlocks$, client.finalizedBlock$]).pipe(
|
||||
map(([best, finalized]) => {
|
||||
if (finalized.hash === hash) return BlockState.Finalized
|
||||
if (finalized.number === number) return BlockState.Pruned
|
||||
if (best.some((b) => b.hash === hash)) return BlockState.Best
|
||||
return BlockState.Fork
|
||||
}),
|
||||
),
|
||||
).pipe(
|
||||
takeWhile(
|
||||
(v) => v !== BlockState.Finalized && v !== BlockState.Pruned,
|
||||
true,
|
||||
),
|
||||
)
|
||||
|
||||
const retryOnStopError = <T>() =>
|
||||
retry<T>({
|
||||
delay(error) {
|
||||
if (error instanceof StopError) {
|
||||
return of(null)
|
||||
}
|
||||
throw error
|
||||
},
|
||||
})
|
||||
|
||||
@@ -146,9 +146,7 @@ export const chainClient$ = state(
|
||||
sinkSuspense(),
|
||||
),
|
||||
)
|
||||
export const chainHead$ = state(
|
||||
chainClient$.pipe(map(({ chainHead }) => chainHead)),
|
||||
)
|
||||
export const client$ = state(chainClient$.pipe(map(({ client }) => client)))
|
||||
|
||||
export const unsafeApi$ = chainClient$.pipeState(
|
||||
map(({ client }) => client.getUnsafeApi()),
|
||||
|
||||
Reference in New Issue
Block a user