ames, +close-flows: combine stale flows

This combines two types of flows that can be closed:

- nacking %watch pleas
- resubscriptions that were not properly corked

pkg/arvo/gen/hood/close-flows.hoon

@@ -1,4 +1,11 @@
-::  |close-flows: sends a %cork to all ships that have nacked a %watch
+::  (TODO: rename) |close-flows: shows how many stale ames flows can be closed
+::                (each ames flow is represented by its first bone)
+::
+::    Run in verbose mode my default, to turn of: |close-flows, =veb |
+::
+::    TODO:
+::    corks all ames flows that are in a stale state due to some
+::    previous unhandled errors
 ::
 ::    It runs in dry mode by default, printing the number of flows that
 ::    can be closed. To actually send the corks, run as:
@@ -7,13 +14,12 @@
 ::
 :-  %say
 |=  $:  [now=@da eny=@uvJ bec=beak]
-        [arg=~ dry-run=?]
+        [arg=~ dry=? veb=?]
     ==
 ::
 =/  peers-map
-  .^  (map ship ?(%alien %known))
-      %ax  /(scot %p p.bec)//(scot %da now)/peers
-  ==
+  .^((map ship ?(%alien %known)) %ax /(scot %p p.bec)//(scot %da now)/peers)
+::
 =/  peers=(list ship)
   %+  murn  ~(tap by peers-map)
   |=  [=ship val=?(%alien %known)]
@@ -24,44 +30,75 @@
     %known  (some ship)
   ==
 ::
-=;  krocs=(list [=ship =bone])
-  helm-ames-kroc+[krocs dry-run]
+=/  agents-map
+  .^  (list [app=term sub-nonce=@ud run-nonce=@t =boat:gall =boar:gall])
+    %gr  (scot %p p.bec)  %base  (scot %da now)  ~
+  ==
 ::
-%-  zing
-%+  turn  peers
-|=  =ship
-=+  .^  =ship-state:ames
-        %ax  /(scot %p p.bec)//(scot %da now)/peers/(scot %p ship)
-    ==
-=/  =peer-state:ames  ?>(?=(%known -.ship-state) +.ship-state)
+|^  ~?  veb  "{<~(wyt in nacks)>} flows from %nacking %watches"
+    ~?  veb  "{<~(wyt in resubs)>} flows from stale resubscriptions"
+    :+  %helm-ames-kroc  dry
+    ~(tap in (~(uni in nacks) resubs))
 ::
-^-  (list [@p @ud])
-%+  murn  ~(tap by rcv.peer-state)
-|=  [=bone *]
-?.  &(=(0 (end 0 bone)) =(1 (end 0 (rsh 0 bone))))
-  ::  not a naxplanation ack bone
+++  resubs
+  =+  peers=(~(gas in *(set ship)) peers)
+  %+  roll  agents-map
+  |=  $:  [app=term next-nonce=@ run-nonce=@t =boat:gall =boar:gall]
+          resubs=(set [ship bone])
+      ==
+  =/  subs  ~(tap in ~(key by boat))
+  ?~  subs  resubs
+  =/  [=wire =dock]    i.subs
+  =/  sub=@ud  (~(got by boar) wire dock)
+  ::  ignores subscription with nonce=0, this is handled by +ap-rake
+  ::  XX TODO handle also here?
   ::
-  ~
-::  by only corking forward flows that have received
-::  a nack we avoid corking the current subscription
+  ?:  =(sub 0)  resubs
+  ::  skip current subscription
+  ::
+  ?.  (lth sub (dec next-nonce))  resubs
+  ?:  !(~(has in peers) p.dock)   resubs
+  =+  .^  =ship-state:ames
+          %ax  /(scot %p p.bec)//(scot %da now)/peers/(scot %p p.dock)
+      ==
+  =/  =peer-state:ames  ?>(?=(%known -.ship-state) +.ship-state)
+  %-  ~(rep by by-duct.ossuary.peer-state)
+  |=  [[=duct =bone] resubs=_resubs]
+   =+  sub-nonce=(scot %ud sub)
+   =/  mod=^wire
+    :*  %gall  %use  app  run-nonce
+        %out  (scot %p p.dock)  q.dock
+        sub-nonce  wire
+    ==
+  ?.  ?&  ?=([* [%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] *] duct)
+          =(mod i.t.duct(i.t.t.t.t.t.t.t sub-nonce))
+      ==
+    resubs
+  (~(put in resubs) [p.dock bone])
 ::
-=+  target=(mix 0b10 bone)
-::  make sure that the nack bone has a forward flow
-::
-?~  duct=(~(get by by-bone.ossuary.peer-state) target)
-  ~
-=;  =wire
-  ?~(wire ~ `[ship target])
-::  TMI
-::
-=>  .(duct `^duct`duct)
-|-  ^-  wire
-?~  duct  ~
-::  inspect the wires in the duct, looking for subscriptions
-::
-?.  ?=([%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] i.duct)
-  $(duct t.duct)
-::  extra security check so we know that the nonce is a number
-::
-?~  (rush i.t.t.t.t.t.t.t.i.duct dem)  ~
-i.duct
+++  nacks
+  %+  roll  peers
+  |=  [=ship nacks=(set [ship bone])]
+  =+  .^  =ship-state:ames
+          %ax  /(scot %p p.bec)//(scot %da now)/peers/(scot %p ship)
+      ==
+  =/  =peer-state:ames  ?>(?=(%known -.ship-state) +.ship-state)
+  ::
+  %+  roll  ~(tap by rcv.peer-state)
+  |=  [[=bone *] nacks=_nacks]
+  ?.  &(=(0 (end 0 bone)) =(1 (end 0 (rsh 0 bone))))
+    ::  not a naxplanation ack bone
+    ::
+    nacks
+  ::  by only corking forward flows that have received
+  ::  a nack we avoid corking the current subscription
+  ::
+  =+  target=(mix 0b10 bone)
+  ::  make sure that the nack bone has a forward flow
+  ::
+  ?~  duct=(~(get by by-bone.ossuary.peer-state) target)
+    nacks
+  ?.  ?=([* [%gall %use sub=@ @ %out @ @ nonce=@ pub=@ *] *] duct)
+    nacks
+  (~(put in nacks) [ship target])
+--

pkg/arvo/lib/hood/helm.hoon

@@ -246,7 +246,7 @@
   (emit %pass /helm %arvo %a %stir '')
 ::
 ++  poke-ames-kroc
-  |=  [krocs=(list [=ship =bone]) dry=?]  =<  abet
+  |=  [dry=? krocs=(list [=ship =bone])]  =<  abet
   ?~  krocs  (flog %text "No %ames flow ready to be close")
   %-  emil    ^-  (list card)
   ::  TODO: extend closing flow overview

pkg/arvo/sys/vane/gall.hoon

@@ -1989,6 +1989,19 @@
       [~ ~]
     [~ ~ atom+!>(u.nonce)]
   ::
+  ?:  ?&  =(%r care)
+          ?=(~ path)
+          =([%$ %da now] coin)
+          =(our ship)
+      ==
+    =|  agents=(list [app=term sub-nonce=@ud run-nonce=@t =boat =boar])
+    :+  ~  ~
+    :-  %noun  !>  ^+  agents
+    %+  roll  ~(tap by yokes.state)
+    |=  [[agent=term =yoke] agents=_agents]
+    :_  agents
+    [agent [sub-nonce run-nonce boat boar]:yoke]
+  ::
   ?.  =(our ship)
     ~
   ?.  =([%$ %da now] coin)