ames: add tool for removing failed %watch flows

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

@@ -0,0 +1,73 @@
+::  |close-flows: sends a %cork to all ships that have nacked a %watch
+::
+::    It runs in dry mode by default, printing the number of flows that
+::    can be closed. To actually send the corks, run as:
+::
+::      |close-flows, =dry-run |
+::
+:-  %say
+|=  $:  [now=@da eny=@uvJ bec=beak]
+        [arg=~ dry-run=?]
+    ==
+::
+=/  peers-map
+  .^  (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)]
+  ?:  =(ship p.bec)
+    ~  ::  this is weird, but we saw it
+  ?-  val
+    %alien  ~
+    %known  (some ship)
+  ==
+::
+=;  krocs=(list [=ship =bone])
+    helm-ames-kroc+[krocs dry-run]
+::
+%-  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)
+=/  forward-flows=(set @)
+  %-  ~(gas in *(set @))
+  %+  turn
+    %+  sort  ~(tap by by-bone.ossuary.peer-state)
+    |=  [[a=@ *] [b=@ *]]  (lth a b)
+  head
+::
+=/  nacks=(list bone)
+  %+  turn
+    (sort ~(tap by rcv.peer-state) |=([[a=@ *] [b=@ *]] (lth a b)))
+  head
+::
+^-  (list [@p @ud])
+%+  murn  nacks
+|=  =bone
+?.  &(=(0 (end 0 bone)) =(1 (end 0 (rsh 0 bone))))
+  ::  not a naxplanation ack bone
+  ::
+  ~
+::  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
+::
+?.  (~(has in forward-flows) target)  ~
+?~  duct=(~(get by by-bone.ossuary.peer-state) target)
+  ~
+=/  =wire  (snag 1 u.duct)
+::  only wires with nonces are subscriptions
+::
+?.  ?=([%gall %use sub=@ @ %out @ @ nonce=@ pub=@ ~] wire)
+  ~
+::  extra security check so we know that the nonce is a number
+::
+?~  (rush i.t.t.t.t.t.t.t.wire dem)  ~
+`[ship target]

pkg/arvo/lib/hood/helm.hoon

@@ -245,6 +245,17 @@
   |=  ~  =<  abet
   (emit %pass /helm %arvo %a %stir '')
 ::
+++  poke-ames-kroc
+  |=  [krocs=(list [=ship =bone]) dry=?]  =<  abet
+  ?~  krocs  (flog %text "No %ames flow ready to be close")
+  %-  emil    ^-  (list card)
+  :-  [%pass /di %arvo %d %flog text+"Closing {<(lent krocs)>} ames flows"]
+  ?:  dry  ~
+  %+  turn  krocs
+  |=  [=ship =bone]
+  ^-  card
+  [%pass /helm %arvo %a %kroc ship bone]
+::
 ++  poke-knob
   |=  [error-tag=@tas level=?(%hush %soft %loud)]  =<  abet
   (emit %pass /helm %arvo %d %knob error-tag level)
@@ -280,6 +291,7 @@
     %helm-ames-sift        =;(f (f !<(_+<.f vase)) poke-ames-sift)
     %helm-ames-verb        =;(f (f !<(_+<.f vase)) poke-ames-verb)
     %helm-ames-wake        =;(f (f !<(_+<.f vase)) poke-ames-wake)
+    %helm-ames-kroc        =;(f (f !<(_+<.f vase)) poke-ames-kroc)
     %helm-atom             =;(f (f !<(_+<.f vase)) poke-atom)
     %helm-automass         =;(f (f !<(_+<.f vase)) poke-automass)
     %helm-cancel-automass  =;(f (f !<(_+<.f vase)) poke-cancel-automass)

pkg/arvo/sys/lull.hoon

@@ -352,6 +352,7 @@
   ::    %heed: track peer's responsiveness; gives %clog if slow
   ::    %jilt: stop tracking peer's responsiveness
   ::    %cork: request to delete message flow
+  ::    %kroc: request to delete message flow on a specific bone
   ::    %plea: request to send message
   ::
   ::    System and Lifecycle Tasks
@@ -370,6 +371,7 @@
         [%heed =ship]
         [%jilt =ship]
         [%cork =ship]
+        [%kroc =ship =bone]
         $>(%plea vane-task)
     ::
         $>(%born vane-task)

pkg/arvo/sys/vane/ames.hoon

@@ -1183,6 +1183,7 @@
       %vega  on-vega:event-core
       %plea  (on-plea:event-core [ship plea]:task)
       %cork  (on-cork:event-core ship.task)
+      %kroc  (on-kroc:event-core ship.task bone.task)
     ==
   ::
   [moves ames-gate]
@@ -1908,6 +1909,31 @@
         =/  rcvr  [ship her-life.channel]
         "cork plea {<sndr rcvr bone=bone vane.plea path.plea>}"
     abet:(on-memo:(make-peer-core peer-state channel) bone plea %plea)
+  ::  +on-kroc: explicitly cork a flow on the provided bone
+  ::    TODO: refactor
+  ::
+  ++  on-kroc
+    |=  [=ship =bone]
+    ^+  event-core
+    =/  =plea       [%$ /flow [%cork ~]]
+    =/  ship-state  (~(get by peers.ames-state) ship)
+    ?.  ?=([~ %known *] ship-state)
+      %+  enqueue-alien-todo  ship
+      |=  todos=alien-agenda
+      todos(messages [[duct plea] messages.todos])
+    =/  =peer-state  +.u.ship-state
+    ?.  (~(has by by-bone.ossuary.peer-state) bone)
+      ~&  >>>  %dont-have-it^ship^bone
+      event-core
+    =/  =channel  [[our ship] now channel-state -.peer-state]
+    ::
+    =.  closing.peer-state  (~(put in closing.peer-state) bone)
+    %-  %^  trace  msg.veb  ship
+        |.  ^-  tape
+        =/  sndr  [our our-life.channel]
+        =/  rcvr  [ship her-life.channel]
+        "cork plea {<sndr^rcvr^bone=bone^vane.plea^path.plea>}"
+    abet:(on-memo:(make-peer-core peer-state channel) bone plea %plea)
   ::  +on-take-wake: receive wakeup or error notification from behn
   ::
   ++  on-take-wake