feat: produce new blocks
This commit is contained in:
Victor Oliva
+160
-50
@@ -1,76 +1,186 @@
|
||||
import { Blockchain } from "@acala-network/chopsticks-core"
|
||||
import { getSyncProvider } from "@polkadot-api/json-rpc-provider-proxy"
|
||||
import { BehaviorSubject } from "rxjs"
|
||||
import { blockHeader } from "@polkadot-api/substrate-bindings"
|
||||
import { JsonRpcProvider } from "polkadot-api/ws-provider/web"
|
||||
|
||||
export const chopsticksInstance$ = new BehaviorSubject<Blockchain | null>(null)
|
||||
|
||||
export const createChopsticksProvider = (endpoint: string) =>
|
||||
getSyncProvider(async () => {
|
||||
const { ChopsticksProvider, setup } = await import(
|
||||
"@acala-network/chopsticks-core"
|
||||
)
|
||||
withChopsticksEnhancer(
|
||||
getSyncProvider(async () => {
|
||||
const { ChopsticksProvider, setup } = await import(
|
||||
"@acala-network/chopsticks-core"
|
||||
)
|
||||
|
||||
chopsticksInstance$.getValue()?.close()
|
||||
const chain = await setup({
|
||||
endpoint,
|
||||
mockSignatureHost: true,
|
||||
})
|
||||
chopsticksInstance$.next(chain)
|
||||
chopsticksInstance$.getValue()?.close()
|
||||
const chain = await setup({
|
||||
endpoint,
|
||||
mockSignatureHost: true,
|
||||
})
|
||||
chopsticksInstance$.next(chain)
|
||||
|
||||
const innerProvider = new ChopsticksProvider(chain)
|
||||
return (onMessage) => {
|
||||
return {
|
||||
send: async (message: string) => {
|
||||
const parsed = JSON.parse(message)
|
||||
const innerProvider = new ChopsticksProvider(chain)
|
||||
return (onMessage) => {
|
||||
return {
|
||||
send: async (message: string) => {
|
||||
const parsed = JSON.parse(message)
|
||||
|
||||
if (parsed.method === "chainHead_v1_follow") {
|
||||
const subscription = await innerProvider.subscribe(
|
||||
"chainHead_v1_followEvent",
|
||||
if (parsed.method === "chainHead_v1_follow") {
|
||||
const subscription = await innerProvider.subscribe(
|
||||
"chainHead_v1_followEvent",
|
||||
parsed.method,
|
||||
parsed.params,
|
||||
(err, result) => {
|
||||
if (err) {
|
||||
console.error(err)
|
||||
return
|
||||
}
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
method: "chainHead_v1_followEvent",
|
||||
params: {
|
||||
subscription,
|
||||
result,
|
||||
},
|
||||
}),
|
||||
)
|
||||
},
|
||||
)
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: parsed.id,
|
||||
result: subscription,
|
||||
}),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
const response = await innerProvider.send(
|
||||
parsed.method,
|
||||
parsed.params,
|
||||
(err, result) => {
|
||||
if (err) {
|
||||
console.error(err)
|
||||
return
|
||||
}
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
method: "chainHead_v1_followEvent",
|
||||
params: {
|
||||
subscription,
|
||||
result,
|
||||
},
|
||||
}),
|
||||
)
|
||||
},
|
||||
)
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: parsed.id,
|
||||
result: subscription,
|
||||
result: response,
|
||||
}),
|
||||
)
|
||||
return
|
||||
}
|
||||
},
|
||||
disconnect: () => {
|
||||
chain?.close()
|
||||
chopsticksInstance$.next(null)
|
||||
},
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
|
||||
const response = await innerProvider.send(
|
||||
parsed.method,
|
||||
parsed.params,
|
||||
)
|
||||
/**
|
||||
* Chopsticks can create block number discontinuities on the chain, which breaks an assumption of polkadot-api.
|
||||
* The spec-compliant way of solving this is by emitting a stop event when that happens
|
||||
*/
|
||||
const withChopsticksEnhancer =
|
||||
(parent: JsonRpcProvider): JsonRpcProvider =>
|
||||
(onMessage) => {
|
||||
// if it's chopsticks, we can assume there's immediate finality, and there are no forks or reorgs
|
||||
let previousNumber: number | null = null
|
||||
let waitingForNumber: any = null
|
||||
const messageQueue: any[] = []
|
||||
|
||||
const processMessage = (parsed: any) => {
|
||||
if (parsed.id?.startsWith("chopsticks-header-")) {
|
||||
const decodedHeader = blockHeader.dec(parsed.result)
|
||||
const currentNumber = decodedHeader.number
|
||||
|
||||
if (
|
||||
waitingForNumber &&
|
||||
previousNumber !== null &&
|
||||
currentNumber > previousNumber + 1
|
||||
) {
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: parsed.id,
|
||||
result: response,
|
||||
...waitingForNumber,
|
||||
params: {
|
||||
...waitingForNumber.params,
|
||||
result: {
|
||||
event: "stop",
|
||||
},
|
||||
},
|
||||
}),
|
||||
)
|
||||
},
|
||||
disconnect: () => {
|
||||
chain?.close()
|
||||
chopsticksInstance$.next(null)
|
||||
},
|
||||
inner.send(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: "chopsticks-stopped",
|
||||
method: "chainHead_v1_unfollow",
|
||||
params: [waitingForNumber.params.subscription],
|
||||
}),
|
||||
)
|
||||
messageQueue.length = 0
|
||||
previousNumber = currentNumber
|
||||
waitingForNumber = null
|
||||
return
|
||||
}
|
||||
previousNumber = currentNumber
|
||||
|
||||
if (waitingForNumber) {
|
||||
onMessage(JSON.stringify(waitingForNumber))
|
||||
waitingForNumber = null
|
||||
}
|
||||
|
||||
if (messageQueue.length) {
|
||||
const [next] = messageQueue.splice(0, 1)
|
||||
processMessage(next)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (waitingForNumber) {
|
||||
messageQueue.push(parsed)
|
||||
return
|
||||
}
|
||||
|
||||
if (
|
||||
parsed.method === "chainHead_v1_followEvent" &&
|
||||
parsed.params?.result?.event === "newBlock"
|
||||
) {
|
||||
const { blockHash } = parsed.params.result
|
||||
waitingForNumber = parsed
|
||||
|
||||
inner.send(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: "chopsticks-header-" + blockHash,
|
||||
method: "chainHead_v1_header",
|
||||
params: [parsed.params.subscription, blockHash],
|
||||
}),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
onMessage(JSON.stringify(parsed))
|
||||
if (messageQueue.length) {
|
||||
const [next] = messageQueue.splice(0, 1)
|
||||
processMessage(next)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
const inner = parent((msg) => {
|
||||
const parsed = JSON.parse(msg)
|
||||
|
||||
processMessage(parsed)
|
||||
})
|
||||
|
||||
return {
|
||||
send(message) {
|
||||
inner.send(message)
|
||||
},
|
||||
disconnect() {
|
||||
inner.disconnect()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,121 +0,0 @@
|
||||
import {
|
||||
JsonRpcConnection,
|
||||
JsonRpcProvider,
|
||||
} from "@polkadot-api/substrate-client"
|
||||
import {
|
||||
connectable,
|
||||
filter,
|
||||
firstValueFrom,
|
||||
Observable,
|
||||
Subject,
|
||||
Subscription,
|
||||
} from "rxjs"
|
||||
// import { Provider } from "@acala-network/chopsticks-core"
|
||||
|
||||
type ProviderInterfaceCallback = (error: Error | null, result: any) => void
|
||||
type ProviderInterfaceEmitted = "connected" | "disconnected" | "error"
|
||||
type ProviderInterfaceEmitCb = (value?: any) => any
|
||||
interface ProviderInterface {
|
||||
/** true if the provider supports subscriptions (not available for HTTP) */
|
||||
readonly hasSubscriptions: boolean
|
||||
/** true if the clone() functionality is available on the provider */
|
||||
readonly isClonable: boolean
|
||||
/** true if the provider is currently connected (ws/sc has connection logic) */
|
||||
readonly isConnected: boolean
|
||||
clone(): ProviderInterface
|
||||
connect(): Promise<void>
|
||||
disconnect(): Promise<void>
|
||||
on(type: ProviderInterfaceEmitted, sub: ProviderInterfaceEmitCb): () => void
|
||||
send<T = any>(
|
||||
method: string,
|
||||
params: unknown[],
|
||||
isCacheable?: boolean,
|
||||
): Promise<T>
|
||||
subscribe(
|
||||
type: string,
|
||||
method: string,
|
||||
params: unknown[],
|
||||
cb: ProviderInterfaceCallback,
|
||||
): Promise<number | string>
|
||||
unsubscribe(
|
||||
type: string,
|
||||
method: string,
|
||||
id: number | string,
|
||||
): Promise<boolean>
|
||||
}
|
||||
|
||||
type JsonRpcResponse = {
|
||||
jsonrpc: "2.0"
|
||||
id: number
|
||||
result?: unknown
|
||||
error?: unknown
|
||||
}
|
||||
type JsonRpcRequest = {
|
||||
jsonrpc: "2.0"
|
||||
method: string
|
||||
params: unknown[]
|
||||
id?: number
|
||||
}
|
||||
|
||||
export const chopsticksProviderInterfaceFromJsonRPC = (
|
||||
provider: JsonRpcProvider,
|
||||
): ProviderInterface => {
|
||||
let connection: JsonRpcConnection | null = null
|
||||
const connectionEvt$ = new Subject<ProviderInterfaceEmitted>()
|
||||
const shared$ = connectable(
|
||||
new Observable<JsonRpcResponse>((obs) => {
|
||||
connection = provider((v) => obs.next(JSON.parse(v)))
|
||||
connectionEvt$.next("connected")
|
||||
|
||||
return () => {
|
||||
connection?.disconnect()
|
||||
connection = null
|
||||
connectionEvt$.next("disconnected")
|
||||
}
|
||||
}),
|
||||
)
|
||||
let topSubscription: Subscription | null = null
|
||||
|
||||
let nextId = 0
|
||||
|
||||
return {
|
||||
hasSubscriptions: true,
|
||||
isClonable: false,
|
||||
get isConnected() {
|
||||
return !!connection
|
||||
},
|
||||
clone() {
|
||||
throw new Error("Not clonable")
|
||||
},
|
||||
async connect() {
|
||||
if (connection) return
|
||||
topSubscription = shared$.connect()
|
||||
},
|
||||
async disconnect() {
|
||||
topSubscription?.unsubscribe()
|
||||
},
|
||||
on(type, cb) {
|
||||
const sub = connectionEvt$.pipe(filter((v) => v === type)).subscribe(cb)
|
||||
return () => sub.unsubscribe()
|
||||
},
|
||||
async send<T>(method: string, params: unknown[]) {
|
||||
const id = nextId++
|
||||
const responsePromise = firstValueFrom(
|
||||
shared$.pipe(filter((v) => v.id === id)),
|
||||
)
|
||||
|
||||
connection?.send(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
method,
|
||||
params,
|
||||
id,
|
||||
} satisfies JsonRpcRequest),
|
||||
)
|
||||
|
||||
const response = await responsePromise
|
||||
if (response.result) return response.result as T
|
||||
throw response.error
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -1,15 +1,19 @@
|
||||
import { chopsticksInstance$ } from "@/chopsticks/chopsticks"
|
||||
import { Chopsticks } from "@/components/Icons"
|
||||
import { Button } from "@/components/ui/button"
|
||||
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 { state, useStateObservable, withDefault } from "@react-rxjs/core"
|
||||
import { FC, PropsWithChildren, ReactElement, useEffect, useState } from "react"
|
||||
import { firstValueFrom, map, switchMap } from "rxjs"
|
||||
import { twMerge } from "tailwind-merge"
|
||||
import { BlockTime } from "./BlockTime"
|
||||
import { EpochRemainingTime } from "./EpochTime"
|
||||
|
||||
const finalized$ = client$.pipeState(
|
||||
const finalizedNum$ = client$.pipeState(
|
||||
switchMap((chainHead) => chainHead.finalizedBlock$),
|
||||
map((v) => v.number.toLocaleString()),
|
||||
map((v) => v.number),
|
||||
)
|
||||
const finalized$ = finalizedNum$.pipeState(map((v) => v.toLocaleString()))
|
||||
const best$ = client$.pipeState(
|
||||
switchMap((chainHead) => chainHead.bestBlocks$),
|
||||
map(([v]) => v.number.toLocaleString()),
|
||||
@@ -27,13 +31,20 @@ const hasEpoch$ = runtimeCtx$.pipeState(
|
||||
withDefault(false),
|
||||
)
|
||||
|
||||
const canJump$ = state(chopsticksInstance$.pipe(map((v) => !!v)), false)
|
||||
export const Summary: FC = () => {
|
||||
const hasEpoch = useStateObservable(hasEpoch$)
|
||||
const canJump = useStateObservable(canJump$)
|
||||
|
||||
return (
|
||||
<div className="flex gap-4 items-center py-2">
|
||||
<SummaryItem title="Block Time" className="bg-card/0 border-none">
|
||||
<BlockTime />
|
||||
</SummaryItem>
|
||||
{canJump ? (
|
||||
<Jump />
|
||||
) : (
|
||||
<SummaryItem title="Block Time" className="bg-card/0 border-none">
|
||||
<BlockTime />
|
||||
</SummaryItem>
|
||||
)}
|
||||
{hasEpoch ? (
|
||||
<SummaryItem title="Epoch" className="bg-card/0 border-none">
|
||||
<EpochRemainingTime />
|
||||
@@ -47,7 +58,7 @@ export const Summary: FC = () => {
|
||||
}
|
||||
|
||||
const SummaryItem: FC<
|
||||
PropsWithChildren<{ title: string; className?: string }>
|
||||
PropsWithChildren<{ title: string | ReactElement; className?: string }>
|
||||
> = ({ title, className, children }) => {
|
||||
return (
|
||||
<div
|
||||
@@ -63,3 +74,51 @@ const SummaryItem: FC<
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
const Jump = () => {
|
||||
const finalized = useStateObservable(finalizedNum$)
|
||||
const [value, setValue] = useState(finalized + 1)
|
||||
const [loading, setLoading] = useState(false)
|
||||
|
||||
useEffect(() => {
|
||||
setValue((v) => Math.max(v, finalized + 1))
|
||||
}, [finalized])
|
||||
|
||||
return (
|
||||
<SummaryItem title="" className="bg-card/0 border-none text-center">
|
||||
<div className="text-left">
|
||||
<span className="text-sm">New Height</span>
|
||||
<input
|
||||
className="block border rounded p-1"
|
||||
type="number"
|
||||
value={value}
|
||||
onChange={(evt) => setValue(evt.target.valueAsNumber)}
|
||||
/>
|
||||
</div>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="secondary"
|
||||
className="py-1 h-auto mt-2"
|
||||
onClick={async () => {
|
||||
setLoading(true)
|
||||
try {
|
||||
const chop = await firstValueFrom(chopsticksInstance$)
|
||||
await chop?.newBlock(
|
||||
value !== finalized + 1
|
||||
? {
|
||||
unsafeBlockHeight: value,
|
||||
}
|
||||
: {},
|
||||
)
|
||||
} finally {
|
||||
setLoading(false)
|
||||
}
|
||||
}}
|
||||
disabled={loading}
|
||||
>
|
||||
New Block{" "}
|
||||
<Chopsticks className="inline-block align-middle ml-1" size={20} />
|
||||
</Button>
|
||||
</SummaryItem>
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user