shrub/pkg/arvo/app/aggregator.hoon
2021-07-03 12:50:38 +02:00

839 lines
22 KiB
Plaintext

:: aggregator: Azimuth L2 roll aggregator
::
:: general flow is as described below, to ensure transactions actually go
:: through once we start sending it out, in the dumbest reasonable way.
::
:: periodic timer fires:
:: if there are no pending l2 txs, do nothing.
:: else kick off tx submission flow:
:: "freeze" pending txs, store alongside nonce, then increment nonce,
:: kick off thread for sending the corresponding l1 tx:
:: if nonce doesn't match on-chain expected nonce, bail.
:: if we can't afford the tx fee, bail.
:: construct, sign, submit the l1 tx.
:: if thread bailed, retry in five minutes.
:: if thread succeeded, retry in five minutes with higher gas price.
:: when retrying, only do so if l2 txs remain in the "frozen" txs group.
:: on %tx diff from naive, remove the matching tx from the frozen group.
::
::TODO questions:
:: - it's a bit weird how we just assume the raw and tx in raw-tx to match...
::
/- *aggregator
/+ azimuth,
naive,
lib=naive-transactions,
default-agent,
ethereum,
dbug,
verb
::
|%
+$ state-0
$: %0
:: pending: the next l2 txs to be sent
:: sending: the l2 txs currently sending/awaiting l2 confirmation
:: finding: raw-tx-hash reverse lookup for sending map
:: history: status of l2 txs by ethereum address
:: next-nonce: next l1 nonce to use
:: next-batch: when then next l2 batch will be sent
:: pre: predicted l2 state
:: flush: flag for deriving predicted state
::
pending=(list pend-tx)
::
$= sending
%+ map l1-tx-pointer
[next-gas-price=@ud txs=(list raw-tx:naive)]
::
finding=(map keccak ?(%confirmed %failed l1-tx-pointer))
history=(jug address:ethereum roller-tx)
next-nonce=(unit @ud)
next-batch=time
pre=^state:naive
flush=?
::
:: pk: private key to send the roll
:: frequency: time to wait between sending batches (TODO fancier)
:: endpoint: ethereum rpc endpoint to use
:: contract: ethereum contract address
:: chain-id: mainnet, ropsten, local (https://chainid.network/)
::
pk=@
frequency=@dr
endpoint=(unit @t)
contract=@ux
chain-id=@
==
::
+$ config
$% [%frequency frequency=@dr]
[%setkey pk=@]
[%endpoint endpoint=@t]
[%network net=?(%mainnet %ropsten %local)]
==
::
+$ action
$% [%submit force=? sig=@ tx=part-tx]
[%cancel sig=@ keccak=@ =l2-tx]
[%commit ~] ::TODO maybe pk=(unit @) later
[%config config]
==
::
+$ card card:agent:gall
::
:: TODO: add to config
::
++ resend-time ~m5
::
++ lverb &
--
::
=| state-0
=* state -
::
%- agent:dbug
%+ verb |
^- agent:gall
::
=<
|_ =bowl:gall
+* this .
do ~(. +> bowl)
def ~(. (default-agent this %|) bowl)
::
++ on-init
^- (quip card _this)
=. frequency ~h1
=. contract naive:local-contracts:azimuth
=. chain-id chain-id:local-contracts:azimuth
=^ card next-batch set-timer
:_ this
:~ card
[%pass /azimuth-txs %agent [our.bowl %azimuth] %watch /txs]
==
::
++ on-save !>(state)
++ on-load
|= old=vase
^- (quip card _this)
[~ this(state !<(state-0 old))]
::
++ on-poke
|= [=mark =vase]
^- (quip card _this)
=^ cards state
?+ mark (on-poke:def mark vase)
%aggregator-action
=+ !<(poke=action vase)
(on-action:do poke)
==
[cards this]
:: +on-peek: scry paths
::
:: /x/pending -> %noun (list pend-tx)
:: /x/pending/[~ship] -> %noun (list pend-tx)
:: /x/pending/[0xadd.ress] -> %noun (list pend-tx)
:: /x/tx/[0xke.ccak]/status -> %noun tx-status
:: /x/nonce/[~ship]/[proxy] -> %noun (unit @)
:: /x/spawned/[~ship] -> %noun (list [ship address])
:: /x/next-batch -> %atom time
:: /x/point/[~ship] -> %noun point:naive
:: /x/points/[0xadd.ress] -> %noun (list [ship point:naive])
:: /x/config -> %noun config
::
++ on-peek
|= =path
^- (unit (unit cage))
|^
?+ path ~
[%x %pending ~] ``noun+!>(pending)
[%x %pending @ ~] (pending-by i.t.t.path)
[%x %tx @ %status ~] (status i.t.t.path)
[%x %history @ ~] (history i.t.t.path)
[%x %nonce @ @ ~] (nonce i.t.t.path i.t.t.t.path)
[%x %spawned @ ~] (spawned i.t.t.path)
[%x %next-batch ~] ``noun+!>(next-batch)
[%x %point @ ~] (point i.t.t.path)
[%x %points @ ~] (points i.t.t.path)
[%x %config ~] config
==
::
++ pending-by
|= wat=@t
?~ who=(slaw %p wat)
:: by-address
::
?~ wer=(slaw %ux wat)
[~ ~]
=; pending=(list pend-tx)
``noun+!>(pending)
%+ skim pending
|= pend-tx
=(u.wer (get-l1-address tx.raw-tx pre))
:: by-ship
::
=; pending=(list pend-tx)
``noun+!>(pending)
%+ skim pending
|= pend-tx
=(u.who ship.from.tx.raw-tx)
::
++ status
|= wat=@t
?~ keccak=(slaw %ux wat)
[~ ~]
:+ ~ ~
:- %noun
!> ^- tx-status
?^ status=(~(get by finding) u.keccak)
?@ u.status [u.status ~]
[%sending status]
::TODO potentially slow!
=; known=?
[?:(known %pending %unknown) ~]
%+ lien pending
|= pend-tx
=(u.keccak (hash-tx raw.raw-tx))
::
++ history
|= wat=@t
:+ ~ ~
:- %noun
!> ^- (list roller-tx)
?~ addr=(slaw %ux wat) ~
%~ tap in
(~(get ju ^history) u.addr)
::
++ nonce
|= [who=@t proxy=@t]
?~ who=(slaw %p who)
[~ ~]
?. ?=(proxy:naive proxy)
[~ ~]
:+ ~ ~
:- %noun
!> ^- (unit @)
?~ point=(get:orm:naive points.pre u.who)
~
=< `nonce
(proxy-from-point:naive proxy u.point)
::
++ spawned
|= wat=@t
:+ ~ ~
:- %noun
!> ^- (list [=^ship =address:ethereum])
?~ star=(slaw %p wat) ~
=/ range
%+ lot:orm:naive points.pre
:: range exclusive [star next-star-first-planet-]
:: TODO: make range inclusive ([first-planet last-planet])?
::
[`u.star `(cat 3 +(u.star) 0x1)]
%+ turn (tap:orm:naive range)
|= [=ship =point:naive]
^- [=^ship =address:ethereum]
:- ship
address:(proxy-from-point:naive %own point)
::
++ point
|= wat=@t
?~ ship=(rush wat ;~(pfix sig fed:ag))
``noun+!>(*(unit point:naive))
``noun+!>((get:orm:naive points.pre u.ship))
::
++ points
|= wat=@t
:+ ~ ~
:- %noun
!> ^- (list [ship point:naive])
?~ addr=(slaw %ux wat)
~
%~ tap in
(~(get ju owners.pre) u.addr)
::
++ config
:+ ~ ~
:- %noun
!> ^- roller-config
:* next-batch
frequency
resend-time
contract
chain-id
==
--
::
++ on-arvo
|= [=wire =sign-arvo]
^- (quip card _this)
?+ wire (on-arvo:def wire sign-arvo)
[%timer ~]
?+ +<.sign-arvo (on-arvo:def wire sign-arvo)
%wake =^(cards state on-timer:do [cards this])
==
::
[%predict ~]
?+ +<.sign-arvo (on-arvo:def wire sign-arvo)
%wake
=. state (predicted-state canonical-state):do
`this(flush &)
==
::
[%resend @ @ ~]
=/ [address=@ux nonce=@ud]
[(slav %ux i.t.wire) (rash i.t.t.wire dem)]
?+ +<.sign-arvo (on-arvo:def wire sign-arvo)
%wake [(send-roll:do address nonce) this]
==
==
::
++ on-fail
|= [=term =tang]
::TODO if crashed during timer, set new timer? how to detect?
(on-fail:def term tang)
::
++ on-watch on-watch:def
++ on-leave on-leave:def
++ on-agent
|= [=wire =sign:agent:gall]
^- (quip card _this)
|^
?+ wire (on-agent:def wire sign)
[%send @ @ *] (send-batch i.t.wire i.t.t.wire sign)
[%azimuth-txs ~] (azimuth-update sign)
[%nonce ~] (nonce sign)
==
::
++ send-batch
|= [address=@t nonce=@t =sign:agent:gall]
^- (quip card _this)
=/ [address=@ux nonce=@ud]
[(slav %ux address) (rash nonce dem)]
?- -.sign
%poke-ack
?~ p.sign
%- (slog leaf+"Send batch thread started successfully" ~)
[~ this]
%- (slog leaf+"{(trip dap.bowl)} couldn't start thread" u.p.sign)
:_ this
[(leave:spider:do wire)]~
::
%watch-ack
?~ p.sign
[~ this]
=/ =tank leaf+"{(trip dap.bowl)} couldn't start listen to thread"
%- (slog tank u.p.sign)
[~ this]
::
%kick
[~ this]
::
%fact
?+ p.cage.sign (on-agent:def wire sign)
%thread-fail
=+ !<([=term =tang] q.cage.sign)
%- (slog leaf+"{(trip dap.bowl)} failed" leaf+<term> tang)
=^ cards state
(on-batch-result:do address nonce %.n^'thread failed')
[cards this]
::
%thread-done
=+ !<(result=(each @ud @t) q.cage.sign)
=^ cards state
(on-batch-result:do address nonce result)
[cards this]
==
==
::
++ azimuth-update
|= =sign:agent:gall
^- (quip card _this)
?+ -.sign [~ this]
%watch-ack
?~ p.sign [~ this]
=/ =tank leaf+"{(trip dap.bowl)} couldn't start listen to %azimuth"
%- (slog tank u.p.sign)
[~ this]
::
%fact
?+ p.cage.sign (on-agent:def wire sign)
%naive-diffs
=+ !<(=diff:naive q.cage.sign)
=^ cards state
(on-naive-diff:do diff)
[cards this]
::
%naive-state
~& > %received-naive-state
:- ~
:: cache naive state, received upon innitializing subscription
:: this assumes that /app/azimuth has already processed eth data
::
=. state (predicted-state:do !<(^state:naive q.cage.sign))
this
==
==
::
++ nonce
|= =sign:agent:gall
^- (quip card _this)
?- -.sign
%poke-ack
?~ p.sign
%- (slog leaf+"Nonce thread started successfully" ~)
[~ this]
%- (slog leaf+"{(trip dap.bowl)} couldn't start thread" u.p.sign)
:_ this
[(leave:spider:do wire)]~
::
%watch-ack
?~ p.sign
[~ this]
=/ =tank leaf+"{(trip dap.bowl)} couldn't start listen to thread"
%- (slog tank u.p.sign)
[~ this]
::
%kick
[~ this]
::
%fact
?+ p.cage.sign (on-agent:def wire sign)
%thread-fail
=+ !<([=term =tang] q.cage.sign)
%- (slog leaf+"{(trip dap.bowl)} failed" leaf+<term> tang)
[~ this]
::
%thread-done
=+ !<(nonce=@ud q.cage.sign)
[~ this(next-nonce `nonce)]
==
==
--
--
::
|_ =bowl:gall
::TODO /lib/sys.hoon?
++ sys
|%
++ b
|%
++ wait
|= [=wire =time]
^- card
[%pass wire %arvo %b %wait time]
--
--
::TODO /lib/spider.hoon?
++ spider
|%
++ start-thread
|= [=wire thread=term arg=vase]
^- (list card)
=/ tid=@ta (rap 3 thread '--' (scot %uv eny.bowl) ~)
=/ args [~ `tid thread arg]
:~ [%pass wire %agent [our.bowl %spider] %watch /thread-result/[tid]]
[%pass wire %agent [our.bowl %spider] %poke %spider-start !>(args)]
==
::
++ leave
|= =path
^- card
[%pass path %agent [our.bowl %spider] %leave ~]
--
::
++ hash-tx keccak-256:keccak:crypto
::
++ hash-raw-tx
|= =raw-tx:naive
^- @ux
(hash-tx raw.raw-tx)
::
++ part-tx-to-full
|= =part-tx
^- [octs tx:naive]
?- -.part-tx
%raw
?~ batch=(parse-raw-tx:naive q.raw.part-tx)
~& %parse-failed
:: TODO: maybe return a unit if parsing fails?
::
!!
[raw tx]:-.u.batch
::
%don [(gen-tx-octs:lib +.part-tx) +.part-tx]
%ful +.part-tx
==
:: +canonical-state: load current l2 state from /app/azimuth
::
++ canonical-state
.^ ^state:naive
%gx
(scot %p our.bowl)
%azimuth
(scot %da now.bowl)
/nas/nas
==
:: +predicted-state
::
:: derives predicted state from pending/sending txs and
:: canonical state, discarding invalid txs in the process.
::
++ predicted-state
|= nas=^state:naive
^+ state
=. pre.state nas
|^
=^ nes state apply-sending
=^ nep state (update-txs pending %pending)
%_ state
sending nes
pending nep
==
::
++ apply-sending
=| valid=_sending
=+ sending=~(tap by sending)
|- ^+ [valid state]
?~ sending [valid state]
::
=* key p.i.sending
=* val q.i.sending
=^ new-valid state
%+ update-txs
(turn txs.val |=(=raw-tx:naive [| 0x0 raw-tx]))
%sending
=. valid
%+ ~(put by valid) key
val(txs (turn new-valid (cork tail tail)))
$(sending t.sending)
::
++ update-txs
|= [txs=(list pend-tx) type=?(%pending %sending)]
=/ valid=_txs ~
=| local=(set keccak)
|- ^+ [valid state]
?~ txs [valid state]
::
=* tx i.txs
=/ hash=@ux (hash-raw-tx raw-tx.tx)
?: (~(has in local) hash)
:: if tx was already seen here, skip
::
$(txs t.txs)
=^ gud=? pre.state
(try-apply pre.state [force raw-tx]:tx)
=? valid gud (snoc valid tx)
=? finding.state !gud
(~(put by finding.state) [hash %failed])
=? history.state !gud
=/ =roller-tx
[[type ~] hash (l2-tx +<.tx.raw-tx.tx)]
%. [address.tx roller-tx(status [%failed ~])]
~(put ju (~(del ju history.state) address.tx roller-tx))
$(txs t.txs, local (~(put in local) hash))
::
++ try-apply
|= [nas=^state:naive force=? =raw-tx:naive]
^- [success=? _nas]
=/ chain-t=@t (ud-to-ascii:naive chain-id)
?. (verify-sig-and-nonce:naive verifier:lib chain-t nas raw-tx)
~& [%verify-sig-and-nonce %failed]
[force nas]
=^ * points.nas
(increment-nonce:naive nas from.tx.raw-tx)
?~ nex=(receive-tx:naive nas tx.raw-tx)
[force nas]
[& +.u.nex]
--
::
++ get-l1-address
|= [=tx:naive nas=^state:naive]
^- address:ethereum
?~ point=(get:orm:naive points.nas ship.from.tx)
!!
=< address
(proxy-from-point:naive proxy.from.tx u.point)
::
++ on-action
|= =action
^- (quip card _state)
?- -.action
%commit on-timer
%config (on-config +.action)
%cancel (cancel-tx +.action)
%submit (take-tx force.action sig.action (part-tx-to-full tx.action))
==
::
++ on-config
|= =config
^- (quip card _state)
?- -.config
%frequency [~ state(frequency frequency.config)]
%endpoint [~ state(endpoint `endpoint.config)]
::
%network
:- ~
=/ [contract=@ux chain-id=@]
=< [naive chain-id]
=, azimuth
?- net.config
%mainnet mainnet-contracts
%ropsten ropsten-contracts
%local local-contracts
==
state(contract contract, chain-id chain-id)
::
%setkey
?~ pk=(de:base16:mimes:html pk.config)
`state
[(get-nonce q.u.pk) state(pk q.u.pk)]
==
:: TODO: move address to state?
::
++ get-address
^- address:ethereum
(address-from-prv:key:ethereum pk)
:: +cancel-tx: cancel a pending transaction
::
++ cancel-tx
|= [sig=@ =keccak =l2-tx]
^- (quip card _state)
?^ status=(~(get by finding) keccak)
~? lverb [dap.bowl %tx-not-pending status+u.status]
[~ state]
:: "cancel: 0x1234abcd"
::
=/ message=octs
%: cad:naive 3
8^'cancel: '
::
=; hash=@t
(met 3 hash)^hash
(crip "0x{((x-co:co 20) keccak)}")
::
~
==
?~ addr=(verify-sig sig message)
~? lverb [dap.bowl %cancel-sig-fail]
[~ state]
=. history
%+ ~(del ju history) u.addr
[[%pending ~] keccak l2-tx]
=. pending
%+ skip pending
|= pend-tx
=(keccak (hash-raw-tx raw-tx))
[~ state]
:: TODO: move to /lib/naive-transactions
::
++ verify-sig
|= [sig=@ txdata=octs]
^- (unit address:naive)
|^
:: Reversed of the usual r-s-v order because Ethereum integers are
:: big-endian
::
=^ v sig (take 3)
=^ s sig (take 3 32)
=^ r sig (take 3 32)
:: In Ethereum, v is generally 27 + recid, and verifier expects a
:: recid. Old versions of geth used 0 + recid, so most software
:: now supports either format. See:
::
:: https://github.com/ethereum/go-ethereum/issues/2053
::
=? v (gte v 27) (sub v 27)
(verifier:lib txdata v r s)
::
++ take
|= =bite
[(end bite sig) (rsh bite sig)]
--
:: +take-tx: accept submitted l2 tx into the :pending list
::
++ take-tx
|= [force=? =raw-tx:naive]
^- (quip card _state)
=/ =address:ethereum
(get-l1-address tx.raw-tx pre)
=/ hash=@ux (hash-raw-tx raw-tx)
:: TODO: what if this hash/tx is already in the history?
:: check in finding that hash doesn't exist ?
::
:: =/ not-sent=? !(~(has by finding) hash)
:: =? pending not-sent
=. pending
(snoc pending [force address raw-tx])
:: =? history not-sent
=. history
%+ ~(put ju history) address
[[%pending ~] hash (l2-tx +<.tx.raw-tx)]
:: ?. not-sent ~& "skip" [~ state]
:: toggle flush flag
::
:_ state(flush ?:(flush | &))
?. flush ~
:: derive predicted state in 5m.
::
[(wait:b:sys /predict (add ~m5 now.bowl))]~
:: +set-timer: %wait until next whole :frequency
::
++ set-timer
^- [=card =time]
=+ time=(mul +((div now.bowl frequency)) frequency)
[(wait:b:sys /timer time) time]
:: +on-timer: every :frequency, freeze :pending txs roll and start sending it
::
++ on-timer
^- (quip card _state)
=. state (predicted-state canonical-state)
=^ cards state
?: =(~ pending) [~ state]
?~ next-nonce
~&([dap.bowl %no-nonce] [~ state])
=/ nonce=@ud u.next-nonce
=: pending ~
flush &
next-nonce `+(u.next-nonce)
::
sending
%+ ~(put by sending)
[get-address nonce]
[0 (turn pending (cork tail tail))]
::
finding
%- ~(gas by finding)
%+ turn pending
|= pend-tx
(hash-raw-tx raw-tx)^[address nonce]
::
history
%+ roll pending
|= [pend-tx hist=_history]
=/ tx=roller-tx
:+ [%pending ~]
(hash-raw-tx raw-tx)
(l2-tx +<.tx.raw-tx)
%+ ~(put ju (~(del ju hist) address tx))
address
tx(status [%sending ~])
==
[(send-roll get-address nonce) state]
=^ card next-batch set-timer
[[card cards] state]
:: +get-nonce: retrieves the latest nonce
::
++ get-nonce
|= pk=@
^- (list card)
?~ endpoint ~&([dap.bowl %no-endpoint] ~)
(start-thread:spider /nonce [%aggregator-nonce !>([u.endpoint pk])])
::
:: +send-roll: start thread to submit roll from :sending to l1
::
++ send-roll
|= [=address:ethereum nonce=@ud]
^- (list card)
:: if this nonce isn't in the sending queue anymore, it's done
::
?. (~(has by sending) [address nonce])
~? lverb [dap.bowl %done-sending [address nonce]]
~
:: start the thread, passing in the l2 txs to use
::
?~ endpoint ~&([dap.bowl %no-endpoint] ~)
::TODO should go ahead and set resend timer in case thread hangs, or nah?
%+ start-thread:spider
/send/(scot %ux address)/(scot %ud nonce)
:- %aggregator-send
!> ^- rpc-send-roll
:* u.endpoint
contract
chain-id
pk
nonce
(~(got by sending) [address nonce])
==
:: +on-batch-result: await resend after thread success or failure
::
++ on-batch-result
|= [=address:ethereum nonce=@ud result=(each @ud @t)]
^- (quip card _state)
:: update gas price for this tx in state
::
=? sending ?=(%& -.result)
%+ ~(jab by sending) [address nonce]
(cork tail (lead p.result))
:: print error if there was one
::
~? ?=(%| -.result) [dap.bowl %send-error p.result]
:: resend the l1 tx in five minutes
::
:_ state
:_ ~
%+ wait:b:sys
/resend/(scot %ux address)/(scot %ud nonce)
(add resend-time now.bowl)
:: +on-naive-diff: process l2 tx confirmations
::
++ on-naive-diff
|= =diff:naive
^- (quip card _state)
?. ?=(%tx -.diff)
[~ state]
=/ =keccak (hash-raw-tx raw-tx.diff)
?~ wer=(~(get by finding) keccak)
[~ state]
:: if we had already seen the tx, no-op
::
?@ u.wer
~? &(?=(%confirmed u.wer) ?=(~ err.diff))
[dap.bowl %weird-double-confirm from.tx.raw-tx.diff]
[~ state]
=* nonce nonce.u.wer
:: remove the tx from the sending map
::
=. sending
?~ sen=(~(get by sending) [get-address nonce])
~& [dap.bowl %weird-double-remove]
sending
?~ nin=(find [raw-tx.diff]~ txs.u.sen)
~& [dap.bowl %weird-unknown]
sending
=. txs.u.sen (oust [u.nin 1] txs.u.sen)
?~ txs.u.sen
~? lverb [dap.bowl %done-with-nonce [get-address nonce]]
(~(del by sending) [get-address nonce])
(~(put by sending) [get-address nonce] u.sen)
:: update the finding map with the new status
::
=. finding
%+ ~(put by finding) keccak
?~ err.diff %confirmed
:: if we kept the forced flag around for longer, we could notify of
:: unexpected tx failures here. would that be useful? probably not?
:: ~? !forced [dap.bowl %aggregated-tx-failed-anyway err.diff]
%failed
::
=. history
=/ tx=roller-tx
:+ [%sending ~]
keccak
(l2-tx +<.tx.raw-tx.diff)
=/ =address:ethereum
(get-l1-address tx.raw-tx.diff pre)
%+ ~(put ju (~(del ju history) address tx))
address
%_ tx
status ?~(err.diff [%confirmed ~] [%failed ~])
==
:_ state(flush ?:(flush | &))
?. flush ~
:: derive predicted state in 5m.
::
[(wait:b:sys /predict (add ~m5 now.bowl))]~
::
--