urbit/shrub
1
1
Fork 0
mirror of https://github.com/urbit/shrub.git synced 2024-12-11 11:02:25 +03:00
Code Issues Packages Projects Releases Wiki Activity

gen: update +stale-flows

Browse Source
This commit is contained in:
yosoyubik 2023-06-08 14:04:13 +02:00
parent 623e0eafb2
commit af4c099ee5
1 changed files with 69 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
pkg/arvo/gen/stale-flows.hoon
View File

@ -4,24 +4,26 @@
:: |stale-flows, =veb %2 :: stale flows that keep (re)trying to connect
:: |stale-flows, =veb %21 :: ... per app (only forward)
:: |stale-flows, =veb %3 :: stale resubscriptions
:: |stale-flows, =veb %4 :: print live naxplanation flows
::
=> |%
+$ subs (jar path [ship bone @])
+$ subs (jar path [ship bone @ close=?])
+$ pags (jar app=term [dst=term =ship =path]) :: per-agent
+$ naks (set [ship bone])
:: verbosity
::
+$ veb ?(%0 %1 %2 %21 %3 %31)
+$ veb ?(%0 %1 %2 %21 %3 %4 ~)
::
++ resubs
|= [=subs =veb]
^- @
::=/ sorted
:: %+ sort ~(tap by subs)
:: |=([a=(list *) b=(list *)] (lte (lent a) (lent b)))
%+ roll ~(tap by subs)::sorted
|= [[k=path v=(list [ship bone @])] num=@]
%+ roll ~(tap by subs)
|= [[k=path v=(list [ship bone @ close=?])] num=@]
=/ in-close=@
(roll v |=([[@ @ @ c=?] n=@] ?:(c +(n) n)))
~? &(=(%3 veb) (gth (lent v) 1))
%+ weld ?: =(in-close 0) ""
"[#{<in-close>} %close] "
"#{<(dec (lent v))>} stale resubs on {<k>}"
?. (gth (lent v) 1) num
(add (dec (lent v)) num)
@ -29,7 +31,7 @@
::
:- %say
|= $: [now=@da eny=@uvJ bec=beak]
[arg=~ dry=? =veb]
[arg=~ peer=(unit @p) dry=? =veb]
==
::
=/ peers-map
@ -45,12 +47,14 @@
%known (some ship)
==
::
=; [[=subs =pags backward=@ forward=@] =naks]
=; [[=subs =pags close=@ incoming=@ outgoing=@ nax=@] =naks]
:- %tang %- flop
%+ weld
:~ leaf+"#{<~(wyt in naks)>} flows from %nacking %watches"
leaf+"#{<backward>} live backward flows with (keep retrying)"
leaf+"#{<forward>} live forward flows with (keep retrying)"
leaf+"#{<incoming>} live backward flows"
leaf+"#{<outgoing>} live forward flows"
leaf+"#{<nax>} live naxplanations"
leaf+"#{<close>} flows in closing"
leaf+"#{<(resubs subs veb)>} stale resubscriptions"
==
?. =(%21 veb) ~
@ -60,13 +64,17 @@
(gth (lent v) (lent w))
|= [app=term v=(list [dst=term =ship =path])]
:- %leaf
%+ weld "#{<(lent v)>} flows for {<app>} with >10 retries"
%+ weld "#{<(lent v)>} flows for {<app>}"
?. =(1 (lent v)) ~
?> ?=(^ v)
" on {<ship.i.v>} to {<dst.i.v>} at {<path.i.v>}"
::
%+ roll peers
|= [=ship [=subs p=pags b=@ f=@] n=naks]
|= [=ship [=subs p=pags cl=@ in=@ ou=@ na=@] =naks]
?: ?& ?=(^ peer)
!=(u.peer ship)
==
+<+
=+ .^ =ship-state:ames
%ax /(scot %p p.bec)//(scot %da now)/peers/(scot %p ship)
==
@ -78,55 +86,63 @@
%+ roll ~(tap by snd.peer-state)
|= $: [=bone message-pump-state:ames]
subs=_subs pags=_p
backward=_b forward=_f
close=_cl
incoming=_in outgoing=_ou nax=_na
==
=, packet-pump-state
=+ closing=(~(has ^in closing.peer-state) bone)
:- ?~ duct=(~(get by by-bone.ossuary.peer-state) bone) subs
?. ?=([* [%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] *] u.duct)
?. ?=([* [%gall %use sub=@ @ %out @ @ *] *] u.duct)
subs
=/ =wire i.t.u.duct
=/ nonce=(unit @) (rush i.t.t.t.t.t.t.t.i.t.u.duct dem)
=/ nonce=(unit @ud)
?~ (slag 7 wire) ~
(slaw %ud &8.wire)
%- ~(add ja subs)
:_ [ship bone ?~(nonce 0 u.nonce)] :: 0, 1?
?~ nonce wire
:_ [ship bone ?~(nonce 0 u.nonce) closing] :: 0 = pre-nonce subscriptions
?~ nonce
wire
:: don't include the sub-nonce in the key
::
(weld (scag 7 wire) (slag 8 wire))
%+ roll ~(tap in live)
|= $: [[msg=@ frag=@] [packet-state:ames *]]
pags=_pags
out=[b=_backward f=_forward]
==
::
?~ live [pags close incoming outgoing nax]
:: only forward flows
::
=? pags &(=(0 (end 0 bone)) (gth tries 10))
=? pags =(%0 (mod bone 4))
?~ duct=(~(get by by-bone.ossuary.peer-state) bone)
pags
?. ?=([* [%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] *] u.duct)
?. ?=([* [%gall %use sub=@ @ %out @ @ *] *] u.duct)
pags
=/ =wire i.t.u.duct
(~(add ja pags) (snag 2 wire) (snag 8 wire) ship (slag 9 wire))
(~(add ja pags) (snag 2 wire) (snag 6 wire) ship (slag 7 wire))
::
~? &(=(%2 veb) (gth tries 10))
=+ arrow=?:(=(0 (end 0 bone)) "<-" "->")
=+ closing=(~(has in closing.peer-state) bone)
%+ weld "{arrow} ({(cite:title ship)}) bone=#{<bone>} "
"closing={<closing>} msg=#{<msg>} frag=#{<frag>} #{<tries>}"
:- pags
=? out (gth tries 10)
?: =(0 (end 0 bone))
[b.out +(f.out)]
[+(b.out) f.out]
out
=? close closing +(close)
~? =(%2 veb)
=/ arrow=tape
?+ (mod bone 4) ~|([%odd-bone bone] !!)
%0 "<-"
%1 "->"
%3 "<-"
==
"{arrow} ({(cite:title ship)}) bone=#{<bone>} closing={<closing>}"
::
=/ is-nax=? =(%3 (mod bone 4))
~? &(=(%4 veb) is-nax)
"nax: ({(cite:title ship)}) bone=#{<bone>} closing={<closing>}"
:+ pags close
?+ (mod bone 4) ~|([%odd-bone bone] !!)
%0 [incoming +(outgoing) nax]
%1 [+(incoming) outgoing nax]
%3 [incoming outgoing +(nax)]
==
::
++ nacks
%+ roll ~(tap by rcv.peer-state)
|= [[=bone *] nacks=_n]
|= [[=bone *] n=_naks]
?. &(=(0 (end 0 bone)) =(1 (end 0 (rsh 0 bone))))
:: not a naxplanation ack bone
::
nacks
n
:: by only corking forward flows that have received
:: a nack we avoid corking the current subscription
::
@ -134,16 +150,22 @@
:: make sure that the nack bone has a forward flow
::
?~ duct=(~(get by by-bone.ossuary.peer-state) target)
nacks
n
?. ?=([* [%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] *] u.duct)
nacks
n
=/ =wire i.t.u.duct
=+ closing=(~(has ^in closing.peer-state) bone)
?> ?=([%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] wire)
=/ app=term i.t.t.wire
=/ nonce=@
=- ?~(- 0 u.-)
(rush i.t.t.t.t.t.t.t.wire dem)
=/ =path t.t.t.t.t.t.t.t.wire
~? =(%1 veb) "[bone={<target>} nonce={<nonce>} agent={<app>}] {<path>}"
(~(put in nacks) [ship target])
(slaw %ud &8.wire)
=/ =path |8.wire
~? =(%1 veb)
"[bone={<target>} nonce={<nonce>} agent={<app>} close={<closing>}] {<path>}"
(~(put ^in n) [ship target])
--