mirror of
https://github.com/urbit/shrub.git
synced 2024-12-19 16:51:42 +03:00
427 lines
13 KiB
Plaintext
427 lines
13 KiB
Plaintext
/+ tapp, stdio
|
|
=, able:jael
|
|
=> |%
|
|
+$ pending-udiffs (map number:block udiffs:point)
|
|
+$ app-state
|
|
$: %2
|
|
url=@ta
|
|
=number:block
|
|
=pending-udiffs
|
|
blocks=(list block)
|
|
whos=(set ship)
|
|
==
|
|
+$ peek-data ~
|
|
+$ in-poke-data
|
|
$: %azimuth-tracker-poke
|
|
$% [%listen whos=(list ship) =source:jael]
|
|
[%watch url=@ta]
|
|
==
|
|
==
|
|
+$ out-poke-data ~
|
|
+$ in-peer-data ~
|
|
+$ out-peer-data
|
|
[%azimuth-udiff =ship =udiff:point]
|
|
++ tapp
|
|
%: ^tapp
|
|
app-state
|
|
peek-data
|
|
in-poke-data
|
|
out-poke-data
|
|
in-peer-data
|
|
out-peer-data
|
|
==
|
|
++ tapp-async tapp-async:tapp
|
|
++ stdio (^stdio out-poke-data out-peer-data)
|
|
--
|
|
::
|
|
:: Async helpers
|
|
::
|
|
=> |%
|
|
++ topics
|
|
|= ships=(set ship)
|
|
^- (list ?(@ux (list @ux)))
|
|
:: The first topic should be one of these event types
|
|
::
|
|
:- => azimuth-events:azimuth
|
|
:~ broke-continuity
|
|
changed-keys
|
|
lost-sponsor
|
|
escape-accepted
|
|
==
|
|
:: If we're looking for a specific set of ships, specify them as
|
|
:: the second topic. Otherwise don't specify the second topic so
|
|
:: we will match all ships.
|
|
::
|
|
?: =(~ ships)
|
|
~
|
|
[(turn ~(tap in ships) ,@) ~]
|
|
::
|
|
++ request-rpc
|
|
|= [url=@ta id=(unit @t) req=request:rpc:ethereum]
|
|
=/ m (async:stdio ,json)
|
|
^- form:m
|
|
%+ (retry json) `10
|
|
=/ m (async:stdio ,(unit json))
|
|
^- form:m
|
|
|^
|
|
=/ =request:http
|
|
:* method=%'POST'
|
|
url=url
|
|
header-list=['Content-Type'^'application/json' ~]
|
|
^= body
|
|
%- some %- as-octt:mimes:html
|
|
%- en-json:html
|
|
(request-to-json:rpc:ethereum id req)
|
|
==
|
|
;< ~ bind:m (send-request:stdio request)
|
|
;< rep=(unit client-response:iris) bind:m
|
|
take-maybe-response:stdio
|
|
?~ rep
|
|
(pure:m ~)
|
|
(parse-response u.rep)
|
|
::
|
|
++ parse-response
|
|
|= =client-response:iris
|
|
=/ m (async:stdio ,(unit json))
|
|
^- form:m
|
|
?> ?=(%finished -.client-response)
|
|
?~ full-file.client-response
|
|
(pure:m ~)
|
|
=/ body=@t q.data.u.full-file.client-response
|
|
=/ jon=(unit json) (de-json:html body)
|
|
?~ jon
|
|
(pure:m ~)
|
|
=, dejs-soft:format
|
|
=/ array=(unit (list response:rpc:jstd))
|
|
((ar parse-one-response) u.jon)
|
|
?~ array
|
|
=/ res=(unit response:rpc:jstd) (parse-one-response u.jon)
|
|
?~ res
|
|
(async-fail:stdio %request-rpc-parse-error >id< ~)
|
|
?: ?=(%error -.u.res)
|
|
(async-fail:stdio %request-rpc-error >id< >+.res< ~)
|
|
?. ?=(%result -.u.res)
|
|
(async-fail:stdio %request-rpc-fail >u.res< ~)
|
|
(pure:m `res.u.res)
|
|
(async-fail:stdio %request-rpc-batch >%not-implemented< ~)
|
|
:: (pure:m `[%batch u.array])
|
|
::
|
|
++ parse-one-response
|
|
|= =json
|
|
^- (unit response:rpc:jstd)
|
|
=/ res=(unit [@t ^json])
|
|
%. json
|
|
=, dejs-soft:format
|
|
(ot id+so result+some ~)
|
|
?^ res `[%result u.res]
|
|
~| parse-one-response=json
|
|
:+ ~ %error %- need
|
|
%. json
|
|
=, dejs-soft:format
|
|
(ot id+so error+(ot code+no message+so ~) ~)
|
|
--
|
|
::
|
|
++ retry
|
|
|* result=mold
|
|
|= [crash-after=(unit @ud) computation=_*form:(async:stdio (unit result))]
|
|
=/ m (async:stdio ,result)
|
|
=| try=@ud
|
|
|^
|
|
|- ^- form:m
|
|
=* loop $
|
|
?: =(crash-after `try)
|
|
(async-fail:stdio %retry-too-many ~)
|
|
;< ~ bind:m (backoff try ~m1)
|
|
;< res=(unit result) bind:m computation
|
|
?^ res
|
|
(pure:m u.res)
|
|
loop(try +(try))
|
|
::
|
|
++ backoff
|
|
|= [try=@ud limit=@dr]
|
|
=/ m (async:stdio ,~)
|
|
^- form:m
|
|
;< eny=@uvJ bind:m get-entropy:stdio
|
|
;< now=@da bind:m get-time:stdio
|
|
%- wait:stdio
|
|
%+ add now
|
|
%+ min limit
|
|
?: =(0 try) ~s0
|
|
%+ add
|
|
(mul ~s1 (bex (dec try)))
|
|
(mul ~s0..0001 (~(rad og eny) 1.000))
|
|
--
|
|
::
|
|
++ get-latest-block
|
|
|= url=@ta
|
|
=/ m (async:stdio ,block)
|
|
^- form:m
|
|
;< =json bind:m (request-rpc url `'block number' %eth-block-number ~)
|
|
(get-block-by-number url (parse-eth-block-number:rpc:ethereum json))
|
|
::
|
|
++ get-block-by-number
|
|
|= [url=@ta =number:block]
|
|
=/ m (async:stdio ,block)
|
|
^- form:m
|
|
|^
|
|
;< =json bind:m
|
|
(request-rpc url `'block by number' %eth-get-block-by-number number |)
|
|
=/ =block (parse-block json)
|
|
?. =(number number.id.block)
|
|
(async-fail:stdio %reorg-detected >number< >block< ~)
|
|
(pure:m block)
|
|
::
|
|
++ parse-block
|
|
|= =json
|
|
^- block
|
|
=< [[&1 &2] |2]
|
|
^- [@ @ @]
|
|
~| json
|
|
%. json
|
|
=, dejs:format
|
|
%- ot
|
|
:~ hash+parse-hex-result:rpc:ethereum
|
|
number+parse-hex-result:rpc:ethereum
|
|
'parentHash'^parse-hex-result:rpc:ethereum
|
|
==
|
|
--
|
|
::
|
|
++ get-logs-by-hash
|
|
|= [url=@ta whos=(set ship) =hash:block]
|
|
=/ m (async:stdio udiffs:point)
|
|
^- form:m
|
|
;< =json bind:m
|
|
%+ request-rpc url
|
|
:* `'logs by hash'
|
|
%eth-get-logs-by-hash
|
|
hash
|
|
~[azimuth:contracts:azimuth]
|
|
(topics whos)
|
|
==
|
|
=/ event-logs=(list event-log:rpc:ethereum)
|
|
(parse-event-logs:rpc:ethereum json)
|
|
=/ =udiffs:point (event-logs-to-udiffs event-logs)
|
|
(pure:m udiffs)
|
|
::
|
|
++ get-logs-by-range
|
|
|= [url=@ta whos=(set ship) =from=number:block =to=number:block]
|
|
=/ m (async:stdio udiffs:point)
|
|
^- form:m
|
|
;< =json bind:m
|
|
%+ request-rpc url
|
|
:* `'logs by range'
|
|
%eth-get-logs
|
|
`number+from-number
|
|
`number+to-number
|
|
~[azimuth:contracts:azimuth]
|
|
(topics whos)
|
|
==
|
|
=/ event-logs=(list event-log:rpc:ethereum)
|
|
(parse-event-logs:rpc:ethereum json)
|
|
=/ =udiffs:point (event-logs-to-udiffs event-logs)
|
|
(pure:m udiffs)
|
|
::
|
|
++ event-logs-to-udiffs
|
|
|= event-logs=(list =event-log:rpc:ethereum)
|
|
^- =udiffs:point
|
|
%+ murn event-logs
|
|
|= =event-log:rpc:ethereum
|
|
^- (unit [=ship =udiff:point])
|
|
?~ mined.event-log
|
|
~
|
|
?: removed.u.mined.event-log
|
|
~& [%removed-log event-log]
|
|
~
|
|
=/ =id:block [block-hash block-number]:u.mined.event-log
|
|
=, azimuth-events:azimuth
|
|
=, abi:ethereum
|
|
?: =(broke-continuity i.topics.event-log)
|
|
=/ who=@ (decode-topics t.topics.event-log ~[%uint])
|
|
=/ num=@ (decode-results data.event-log ~[%uint])
|
|
`[who id %rift num]
|
|
?: =(changed-keys i.topics.event-log)
|
|
=/ who=@ (decode-topics t.topics.event-log ~[%uint])
|
|
=+ ^- [enc=octs aut=octs sut=@ud rev=@ud]
|
|
%+ decode-results data.event-log
|
|
~[[%bytes-n 32] [%bytes-n 32] %uint %uint]
|
|
`[who id %keys rev sut (pass-from-eth:azimuth enc aut sut)]
|
|
?: =(lost-sponsor i.topics.event-log)
|
|
=+ ^- [who=@ pos=@]
|
|
(decode-topics t.topics.event-log ~[%uint %uint])
|
|
`[who id %spon ~]
|
|
?: =(escape-accepted i.topics.event-log)
|
|
=+ ^- [who=@ wer=@]
|
|
(decode-topics t.topics.event-log ~[%uint %uint])
|
|
`[who id %spon `wer]
|
|
~& [%bad-topic event-log]
|
|
~
|
|
::
|
|
++ jael-update
|
|
|= =udiffs:point
|
|
=/ m (async:stdio ,~)
|
|
|- ^- form:m
|
|
=* loop $
|
|
?~ udiffs
|
|
(pure:m ~)
|
|
=/ =path /(scot %p ship.i.udiffs)
|
|
;< ~ bind:m (give-result:stdio / %azimuth-udiff i.udiffs)
|
|
;< ~ bind:m (give-result:stdio path %azimuth-udiff i.udiffs)
|
|
loop(udiffs t.udiffs)
|
|
--
|
|
::
|
|
:: Main loop
|
|
::
|
|
=> |%
|
|
::
|
|
:: Send %listen to jael
|
|
::
|
|
++ listen
|
|
|= [state=app-state whos=(list ship) =source:jael]
|
|
=/ m (async:stdio ,app-state)
|
|
^- form:m
|
|
;< ~ bind:m (send-effect:stdio %listen /lo (silt whos) source)
|
|
(pure:m state)
|
|
::
|
|
:: Start watching a node
|
|
::
|
|
++ start
|
|
|= state=app-state
|
|
=/ m (async:stdio ,app-state)
|
|
^- form:m
|
|
=: number.state 0
|
|
pending-udiffs.state *pending-udiffs
|
|
blocks.state *(list block)
|
|
==
|
|
(get-updates state)
|
|
::
|
|
:: Get updates since last checked
|
|
::
|
|
++ get-updates
|
|
|= state=app-state
|
|
=/ m (async:stdio ,app-state)
|
|
^- form:m
|
|
;< =latest=block bind:m (get-latest-block url.state)
|
|
;< state=app-state bind:m (zoom state number.id.latest-block)
|
|
|- ^- form:m
|
|
=* walk-loop $
|
|
?: (gth number.state number.id.latest-block)
|
|
;< now=@da bind:m get-time:stdio
|
|
;< ~ bind:m (wait-effect:stdio (add now ~s5))
|
|
(pure:m state)
|
|
;< =block bind:m (get-block-by-number url.state number.state)
|
|
;< [=new=pending-udiffs new-blocks=(lest ^block)] bind:m
|
|
%- take-block
|
|
[url.state whos.state pending-udiffs.state block blocks.state]
|
|
=: pending-udiffs.state new-pending-udiffs
|
|
blocks.state new-blocks
|
|
number.state +(number.id.i.new-blocks)
|
|
==
|
|
walk-loop
|
|
::
|
|
:: Process a block, detecting and handling reorgs
|
|
::
|
|
++ take-block
|
|
|= [url=@ta whos=(set ship) =a=pending-udiffs =block blocks=(list block)]
|
|
=/ m (async:stdio ,[pending-udiffs (lest ^block)])
|
|
^- form:m
|
|
?: &(?=(^ blocks) !=(parent-hash.block hash.id.i.blocks))
|
|
(rewind url a-pending-udiffs block blocks)
|
|
;< =b=pending-udiffs bind:m
|
|
(release-old-events a-pending-udiffs number.id.block)
|
|
;< =new=udiffs:point bind:m (get-logs-by-hash url whos hash.id.block)
|
|
=. b-pending-udiffs (~(put by b-pending-udiffs) number.id.block new-udiffs)
|
|
(pure:m b-pending-udiffs block blocks)
|
|
::
|
|
:: Release events if they're more than 30 blocks ago
|
|
::
|
|
++ release-old-events
|
|
|= [=pending-udiffs =number:block]
|
|
=/ m (async:stdio ,^pending-udiffs)
|
|
^- form:m
|
|
=/ rel-number (sub number 30)
|
|
=/ =udiffs:point (~(get ja pending-udiffs) rel-number)
|
|
;< ~ bind:m (jael-update udiffs)
|
|
(pure:m (~(del by pending-udiffs) rel-number))
|
|
::
|
|
:: Reorg detected, so rewind until we're back in sync
|
|
::
|
|
++ rewind
|
|
|= [url=@ta =pending-udiffs =block blocks=(list block)]
|
|
=/ m (async:stdio ,[^pending-udiffs (lest ^block)])
|
|
|- ^- form:m
|
|
=* loop $
|
|
?~ blocks
|
|
(pure:m pending-udiffs block blocks)
|
|
?: =(parent-hash.block hash.id.i.blocks)
|
|
(pure:m pending-udiffs block blocks)
|
|
;< =next=^block bind:m (get-block-by-number url number.id.i.blocks)
|
|
?: =(~ pending-udiffs)
|
|
;< ~ bind:m (disavow block)
|
|
loop(block next-block, blocks t.blocks)
|
|
=. pending-udiffs (~(del by pending-udiffs) number.id.block)
|
|
loop(block next-block, blocks t.blocks)
|
|
::
|
|
:: Tell subscribers there was a deep reorg
|
|
::
|
|
++ disavow
|
|
|= =block
|
|
=/ m (async:stdio ,~)
|
|
^- form:m
|
|
(jael-update [*ship id.block %disavow ~]~)
|
|
::
|
|
:: Zoom forward to near a given block number.
|
|
::
|
|
:: Zooming doesn't go forward one block at a time. As a
|
|
:: consequence, it cannot detect and handle reorgs. Only use it
|
|
:: at a safe distance -- 500 blocks ago is probably sufficient.
|
|
::
|
|
++ zoom
|
|
|= [state=app-state =latest=number:block]
|
|
=/ m (async:stdio ,app-state)
|
|
^- form:m
|
|
=/ zoom-margin=number:block 100
|
|
?: (lth latest-number (add number.state zoom-margin))
|
|
(pure:m state)
|
|
=/ to-number=number:block (sub latest-number zoom-margin)
|
|
;< =udiffs:point bind:m
|
|
(get-logs-by-range url.state whos.state number.state to-number)
|
|
;< ~ bind:m (jael-update udiffs)
|
|
=. number.state +(to-number)
|
|
=. blocks.state ~
|
|
(pure:m state)
|
|
--
|
|
::
|
|
:: Main
|
|
::
|
|
=* default-tapp default-tapp:tapp
|
|
%- create-tapp-poke-peer-take:tapp
|
|
|_ [=bowl:gall state=app-state]
|
|
++ handle-poke
|
|
|= =in-poke-data
|
|
=/ m tapp-async
|
|
^- form:m
|
|
?- +<.in-poke-data
|
|
%listen (listen state +>.in-poke-data)
|
|
%watch (pure:m state(url url.in-poke-data))
|
|
==
|
|
::
|
|
++ handle-take
|
|
|= =sign:tapp
|
|
=/ m tapp-async
|
|
^- form:m
|
|
?+ -.sign ~|([%strange-sign -.sign] !!)
|
|
%wake (get-updates state)
|
|
==
|
|
::
|
|
++ handle-peer
|
|
|= =path
|
|
=/ m tapp-async
|
|
^- form:m
|
|
=/ who=(unit ship) ?~(path ~ `(slav %p i.path))
|
|
=. whos.state
|
|
?~ who
|
|
~
|
|
(~(put in whos.state) u.who)
|
|
(start state)
|
|
--
|