refchannotify log

docs/todo/misc.txt

@@ -24,3 +24,14 @@ TODO: git-hbs2-export-no-params
    - выполнить export
    - создать конфиг, если его нет
    - добавить ключ в конфиг, если конфиг есть
+
+TODO: metrics-instead-of-plain-errors
+  Сейчас запрос несуществующией ссылки порождает
+  ошибку в логе.
+
+  Иногда это действительно ошибочная ситуация, иногда нет.
+
+  Что бы не засорять лог, лучше считать такие ошибки (запросы
+  к несуществующим ссылкам/блокам), и хранить в таблице,
+  а не писать в лог.
+

hbs2-core/lib/HBS2/Data/Types/Refs.hs

@@ -14,11 +14,10 @@ import Codec.Serialise(serialise)
 import Data.Data
 
 newtype HashRef = HashRef { fromHashRef :: Hash HbSync }
-                  deriving newtype (Eq,Ord,IsString,Pretty,Hashable)
+                  deriving newtype (Eq,Ord,IsString,Pretty,Hashable,Hashed HbSync)
                   deriving stock (Data,Generic,Show)
 
 
-
 instance Pretty (AsBase58 HashRef) where
   pretty (AsBase58 x) = pretty x
   -- TODO: should be instance Pretty (AsBase58 (Hash HbSync))

hbs2-core/lib/HBS2/Hash.hs

@@ -55,6 +55,7 @@ newtype Internal a = Internal a
 class Hashed t a where
   hashObject :: a -> Hash t
 
+deriving newtype instance Hashed HbSync (Hash HbSync)
 
 instance Hashed HbSync ByteString where
   hashObject s = HbSyncHash $! BA.convert digest

hbs2-core/lib/HBS2/Net/Proto/RefChan.hs

@@ -56,6 +56,7 @@ type RefChanAuthor e = PubKey 'Sign (Encryption e)
 
 type Weight = Integer
 
+
 data RefChanHeadBlock e =
   RefChanHeadBlockSmall
    { _refChanHeadVersion     :: Integer
@@ -90,6 +91,8 @@ type ForRefChans e = ( Serialise ( PubKey 'Sign (Encryption e))
                      , Hashable (PubKey 'Sign (Encryption e))
                      )
 
+
+
 refChanHeadReaders :: ForRefChans e => Lens (RefChanHeadBlock e)
                            (RefChanHeadBlock e)
                            (HashSet (PubKey 'Encrypt (Encryption e)))
@@ -815,7 +818,24 @@ refChanRequestProto self adapter msg = do
     proto = Proxy @(RefChanRequest e)
 
 
--- instance Coercible (SignedBox L4Proto
+newtype instance EventKey e (RefChanNotify e) =
+  RefChanNotifyEventKey (RefChanId e)
+
+deriving stock instance ForRefChans e => Typeable (EventKey e (RefChanNotify e))
+deriving stock instance ForRefChans e => Eq  (EventKey e (RefChanNotify e))
+deriving newtype instance ForRefChans e => Hashable  (EventKey e (RefChanNotify e))
+
+data instance Event e (RefChanNotify e) =
+  RefChanNotifyEvent HashRef (RefChanNotify e)
+
+-- FIXME: remove-default-instance?
+instance EventType (Event e (RefChanNotify e)) where
+  isPersistent = True
+
+instance Expires (EventKey e (RefChanNotify e)) where
+  expiresIn = const Nothing -- (Just defCookieTimeoutSec)
+
+
 
 refChanNotifyProto :: forall e s m . ( MonadIO m
                                      , Request e (RefChanNotify e) m
@@ -827,6 +847,7 @@ refChanNotifyProto :: forall e s m . ( MonadIO m
                                      , Pretty (Peer e)
                                      , Sessions e (RefChanHeadBlock e) m
                                      , Sessions e (KnownPeer e) m
+                                     , EventEmitter e (RefChanNotify e) m
                                      , HasStorage m
                                      , Signatures s
                                      , IsRefPubKey s
@@ -892,6 +913,9 @@ refChanNotifyProto self adapter msg@(Notify rchan box) = do
         debug $ "^^^ CALL refChanNotifyRely" <+> pretty h0
         lift $ refChanNotifyRely adapter rchan msg
 
+        debug $ "FUCKING EMIT RefChanNotifyEventKey" <+> pretty (AsBase58 rchan)
+        lift $ emit @e (RefChanNotifyEventKey rchan) (RefChanNotifyEvent (HashRef h0) msg)
+
   where
     proto = Proxy @(RefChanNotify e)
 

hbs2-peer/app/Brains.hs

@@ -38,7 +38,7 @@ import Data.HashMap.Strict qualified as HashMap
 import Data.List qualified as List
 import Data.Maybe
 import Data.Text qualified as Text
-import Data.Time.Clock (UTCTime)
+import Data.Time.Clock (addUTCTime,getCurrentTime)
 import Data.Word
 import Lens.Micro.Platform
 import System.Directory
@@ -250,6 +250,25 @@ instance ( Hashable (Peer e)
       |] ( Only ( show $ pretty (AsBase58 ref) ) )
         <&> isJust . listToMaybe
 
+  setSeen brains w ts = do
+    utc <- liftIO getCurrentTime <&> addUTCTime ts
+    let h = show $ pretty $ hashObject @HbSync w
+    liftIO do
+      let conn = view brainsDb brains
+      void $ execute conn [qc|
+        insert into seen (hash,ts)
+        values (?,?)
+        on conflict (hash) do update set ts = excluded.ts
+      |] (h,utc)
+    commitNow brains False
+
+  isSeen brains w = do
+    let h = show $ pretty $ hashObject @HbSync w
+    liftIO do
+      let conn = view brainsDb brains
+      r <- query @_ @(Only Int) conn [qc|select 1 from seen where hash = ? limit 1|] (Only h)
+      pure $ not $ List.null r
+
 commitNow :: forall e m . MonadIO m
           => BasicBrains e
           -> Bool
@@ -549,6 +568,7 @@ DELETE FROM ancestors WHERE strftime('%s','now') - strftime('%s', ts) > 600;
 DELETE FROM seenby WHERE strftime('%s','now') - strftime('%s', ts) > 600;
 DELETE FROM blocksize WHERE strftime('%s','now') - strftime('%s', ts) > 300;
 DELETE FROM statedb.pexinfo where seen <  datetime('now', '-7 days');
+DELETE FROM seen where ts < datetime('now');
 
 RELEASE SAVEPOINT zzz1;
 
@@ -812,6 +832,14 @@ newBasicBrains cfg = liftIO do
     )
   |]
 
+  execute_ conn [qc|
+    create table if not exists seen
+    ( hash text not null
+    , ts not null
+    , primary key (hash)
+    )
+  |]
+
   BasicBrains <$> newTVarIO mempty
               <*> newTVarIO mempty
               <*> Cache.newCache (Just (toTimeSpec (30 :: Timeout 'Seconds)))

hbs2-peer/app/CLI/RefChan.hs

@@ -8,10 +8,16 @@ import HBS2.Net.Proto.RefChan
 import HBS2.Net.Proto.Types
 import HBS2.Data.Types.SignedBox
 import HBS2.Net.Proto.Service
+import HBS2.Data.Detect
+import HBS2.Storage
+import HBS2.Data.Types.Refs
 
 import HBS2.OrDie
 
 import HBS2.Peer.RPC.API.RefChan
+import HBS2.Peer.RPC.API.Storage
+import HBS2.Peer.RPC.Client.StorageClient
+import HBS2.Peer.RefChanNotifyLog
 
 import CLI.Common
 import RPC2()
@@ -20,9 +26,9 @@ import Options.Applicative
 import Data.ByteString qualified as BS
 import Data.ByteString.Lazy qualified as LBS
 import Lens.Micro.Platform
-import Codec.Serialise
 import Data.Maybe
 import System.Exit
+import Control.Monad.Trans.Maybe
 
 pRefChan :: Parser (IO ())
 pRefChan = hsubparser (   command "head" (info pRefChanHead (progDesc "head commands" ))
@@ -109,10 +115,9 @@ pRefChanPropose = do
   kra <- strOption (long "author" <> short 'a' <> help "author credentials")
   fn  <- optional $ strOption (long "file" <> short 'f' <> help "file")
   dry <- optional (flag' True (long "dry" <> short 'n' <> help "only dump transaction")) <&> fromMaybe False
-  sref <- strArgument (metavar "REFCHAH-KEY")
+  puk <- argument pRefChanId (metavar "REFCHAH-KEY")
   pure $ withMyRPC @RefChanAPI opts $ \caller -> do
     sc <- BS.readFile kra
-    puk <- pure (fromStringMay @(RefChanId L4Proto) sref) `orDie` "can't parse refchan/public key"
     creds <- pure (parseCredentials @(Encryption L4Proto) (AsCredFile sc)) `orDie` "bad keyring file"
 
     lbs <- maybe1 fn LBS.getContents LBS.readFile
@@ -125,20 +130,60 @@ pRefChanPropose = do
       -- FIXME: proper-error-handling
       void $ callService @RpcRefChanPropose caller (puk, box)
 
+
 pRefChanNotify :: Parser (IO ())
-pRefChanNotify = do
+pRefChanNotify =
+  hsubparser (  command  "post"  (info pRefChanNotifyPost    (progDesc "post notify message"))
+             <> command  "log"   (info pRefChanNotifyLog     (progDesc  "post notify message"))
+             <> command  "tail"  (info pRefChanNotifyLogTail (progDesc  "output last messages"))
+             )
+
+pRefChanNotifyPost :: Parser (IO ())
+pRefChanNotifyPost = do
   opts <- pRpcCommon
   kra <- strOption (long "author" <> short 'a' <> help "author credentials")
   fn  <- optional $ strOption (long "file" <> short 'f' <> help "file")
-  sref <- strArgument (metavar "REFCHAH-REF")
+  puk <- argument pRefChanId (metavar "REFCHAH-REF")
   pure $ withMyRPC @RefChanAPI opts $ \caller -> do
     sc <- BS.readFile kra
-    puk <- pure (fromStringMay @(RefChanId L4Proto) sref) `orDie` "can't parse refchan/public key"
     creds <- pure (parseCredentials @(Encryption L4Proto) (AsCredFile sc)) `orDie` "bad keyring file"
     lbs <- maybe1 fn LBS.getContents LBS.readFile
     let box = makeSignedBox @L4Proto @BS.ByteString (view peerSignPk creds) (view peerSignSk creds) (LBS.toStrict lbs)
     void $ callService @RpcRefChanNotify caller (puk, box)
 
+
+pRefChanId :: ReadM (RefChanId L4Proto)
+pRefChanId = maybeReader (fromStringMay @(RefChanId L4Proto))
+
+pRefChanNotifyLog :: Parser (IO ())
+pRefChanNotifyLog = do
+  opts <- pRpcCommon
+  puk <- argument pRefChanId (metavar "REFCHAH-REF")
+  pure $ withMyRPC @StorageAPI opts $ \caller -> do
+    let sto = AnyStorage (StorageClient caller)
+    href <- getRef sto (refAlias (makeRefChanNotifyLogKey @L4Proto puk)) <&> fmap HashRef
+    maybe1 href exitFailure $ \r -> do
+      print $ pretty r
+      exitSuccess
+
+pRefChanNotifyLogTail :: Parser (IO ())
+pRefChanNotifyLogTail = do
+  opts <- pRpcCommon
+
+  n <- optional (option auto ( short 'n' <> long "lines" <> help "output last NUM lines" ))
+          <&> fromMaybe 10
+
+  puk <- argument pRefChanId (metavar "REFCHAH-REF")
+  pure $ withMyRPC @StorageAPI opts $ \caller -> void $ runMaybeT do
+    let sto = AnyStorage (StorageClient caller)
+    href <- getRef sto (refAlias (makeRefChanNotifyLogKey @L4Proto puk)) <&> fmap HashRef
+              >>= toMPlus
+
+    rs <- readLog (getBlock sto) href
+             <&> reverse . take n . reverse
+
+    liftIO $ print $ vcat (fmap pretty rs)
+
 pRefChanGet :: Parser (IO ())
 pRefChanGet = do
   opts <- pRpcCommon

hbs2-peer/app/PeerMain.hs

@@ -59,6 +59,7 @@ import PeerMeta
 import CLI.Common
 import CLI.RefChan
 import RefChan
+import RefChanNotifyLog
 import Fetch (fetchHash)
 import Log
 
@@ -986,6 +987,8 @@ runPeer opts = Exception.handle (\e -> myException e
 
                 peerThread "refChanWorker" (refChanWorker @e rce (SomeBrains brains))
 
+                peerThread "refChanNotifyLogWorker" (refChanNotifyLogWorker @e conf (SomeBrains brains))
+
                 peerThread "all protos" do
                   runProto @e
                     [ makeResponse (blockSizeProto blk dontHandle onNoBlock)

hbs2-peer/app/RefChanNotifyLog.hs

@@ -0,0 +1,154 @@
+{-# LANGUAGE AllowAmbiguousTypes #-}
+{-# Language UndecidableInstances #-}
+module RefChanNotifyLog
+  ( refChanNotifyLogWorker
+  ) where
+
+import HBS2.Prelude.Plated
+
+import HBS2.Actors.Peer
+import HBS2.Hash
+import HBS2.Base58
+import HBS2.Clock
+import HBS2.Events
+import HBS2.Polling
+import HBS2.Data.Detect
+import HBS2.Merkle
+import HBS2.Storage
+import HBS2.Data.Types.Refs
+import HBS2.Net.Auth.Credentials
+import HBS2.Net.Messaging.Unix
+import HBS2.Net.Proto.Definition()
+import HBS2.Net.Proto.Peer
+import HBS2.Net.Proto.RefChan
+import HBS2.Net.Proto.Sessions
+
+import HBS2.System.Logger.Simple
+
+import HBS2.Peer.RefChanNotifyLog
+
+import PeerTypes hiding (downloads)
+import PeerConfig
+import Brains
+
+import Data.Time.Clock (NominalDiffTime(..))
+import Data.List qualified as List
+import Control.Concurrent.STM (flushTQueue)
+
+import Data.Hashable
+import Control.Exception ()
+import Control.Monad
+import Control.Monad.Except ()
+import Data.HashMap.Strict qualified as HashMap
+import Data.Maybe
+import Data.Text qualified as Text
+import UnliftIO
+import Control.Monad.Trans.Maybe
+import Data.Function (on)
+
+import Streaming()
+
+
+data ToListen e = ToListen
+  { listenChan      :: RefChanId e
+  , listenRefKey    :: SomeRefKey (String, RefChanId e)
+  , listenWriteTime :: NominalDiffTime
+  , listenTrim      :: Int
+  }
+
+data MyPoll e = MyPoll (RefChanId e) (ToListen e) (TQueue HashRef)
+
+instance ForRefChans e => Eq (MyPoll e) where
+  (==) (MyPoll a _ _) (MyPoll b _ _) = a == b
+
+instance ForRefChans e => Hashable (MyPoll e) where
+  hashWithSalt salt (MyPoll a _ _) = hashWithSalt salt a
+
+refChanNotifyLogWorker :: forall e s m .
+                          ( MonadIO m
+                          , MonadUnliftIO m
+                          , MyPeer e
+                          , HasStorage m
+                          , Sessions e (KnownPeer e) m
+                          , Signatures s
+                          , s ~ Encryption e
+                          , IsRefPubKey s
+                          , ForRefChans e
+                          , EventListener e (RefChanNotify e) m
+                          )
+             => PeerConfig
+             -> SomeBrains e
+             -> m ()
+
+refChanNotifyLogWorker conf brains = do
+
+  sto <- getStorage
+
+  let refchans = parseConf conf
+
+  qs <- for refchans $ \l -> do
+    let r = listenChan @e l
+
+    q <- newTQueueIO
+
+    subscribe @e (RefChanNotifyEventKey r) $ \(RefChanNotifyEvent h _) -> do
+      seen <- isSeen @e brains h
+      unless seen do
+        trace $ "GOT REFCHAN_NOTIFY TX!" <+> pretty h
+        atomically $ writeTQueue q h
+        -- FIXME: time-hardcode
+        setSeen @e brains h 86400
+
+    pure (l, q)
+
+  polling (Polling 1 1) (toPolling qs) $ \(MyPoll r l q) -> do
+    xs <- atomically $ flushTQueue q
+
+    unless (List.null xs) do
+
+      let ref = listenRefKey l
+
+      v <- getRef sto ref <&> fmap HashRef
+      hashes <- maybe1 v (pure mempty) (readLog (getBlock sto))
+
+      -- TODO: ACTUALLY-MAKE-IT-NOT-SLOW
+      -- TODO: faster-log-merge
+      let newHashes = List.nub $ reverse $ take (listenTrim l) (reverse (hashes <> xs))
+
+      let pt = toPTree (MaxSize 512) (MaxNum 512) newHashes
+
+      v1 <- makeMerkle 0 pt $ \(_,_,bss) -> do
+              void $ putBlock sto bss
+
+      updateRef sto ref v1
+
+      debug $ "REFCHAN_NOTIFY_LOG:" <+> pretty (AsBase58 r)
+                                    <+> pretty (hashObject @HbSync ref)
+                                    <+> pretty v1
+                                    <+> pretty (length newHashes)
+
+  where
+
+    parseConf (PeerConfig syn)  = rcs
+      where rcs = [ ToListen <$> getRefChan rc
+                             <*> (makeRefChanNotifyLogKey @e <$> getRefChan rc)
+                             <*> getWriteTime args
+                             <*> getTrim args
+
+                  | ListVal (  SymbolVal "refchan-notify-log"
+                             : LitStrVal rc
+                             : args
+                           ) <- syn
+                  ] & catMaybes
+                    & List.nubBy ( (==) `on` listenChan )
+
+    getRefChan rc = fromStringMay @(RefChanId e) (Text.unpack rc)
+
+    getWriteTime syn = Just $
+      headDef 1 [ fromIntegral n | ListVal [SymbolVal "write-time", LitIntVal n] <- syn ]
+
+    getTrim syn = Just $
+      headDef 0 [ fromIntegral n | ListVal [SymbolVal "trim", LitIntVal n] <- syn ]
+
+    toPolling qs = pure $ fmap (\(l,q) -> (MyPoll (listenChan l) l q, listenWriteTime l)) qs
+

hbs2-peer/hbs2-peer.cabal

@@ -136,6 +136,7 @@ library
     exposed-modules:
       HBS2.Peer.Brains
       HBS2.Peer.Notify
+      HBS2.Peer.RefChanNotifyLog
       HBS2.Peer.RPC.Class
       HBS2.Peer.RPC.API.Peer
       HBS2.Peer.RPC.API.RefLog
@@ -188,6 +189,7 @@ executable hbs2-peer
                     , PeerConfig
                     , RefLog
                     , RefChan
+                    , RefChanNotifyLog
                     , CheckMetrics
                     , HttpWorker
                     , Brains

hbs2-peer/lib/HBS2/Peer/Brains.hs

@@ -10,6 +10,8 @@ import HBS2.Hash
 import Data.Word
 import HBS2.Data.Types.Refs (HashRef(..))
 
+import Data.Time.Clock (NominalDiffTime)
+
 -- TODO: rename
 class HasBrains e a where
 
@@ -143,6 +145,11 @@ class HasBrains e a where
 
   setReflogProcessed _ _ = pure ()
 
+  setSeen :: (Hashed HbSync x, MonadIO m) => a -> x -> NominalDiffTime -> m ()
+  setSeen _ _ _ = pure ()
+
+  isSeen :: (Hashed HbSync x, MonadIO m) => a -> x -> m Bool
+  isSeen _ _ = pure False
 
 type NoBrains = ()
 
@@ -179,4 +186,6 @@ instance HasBrains e (SomeBrains e) where
   isReflogProcessed (SomeBrains a) = isReflogProcessed @e a
   setReflogProcessed (SomeBrains a) = setReflogProcessed @e a
 
+  setSeen (SomeBrains a) = setSeen @e a
+  isSeen (SomeBrains a) = isSeen @e a
 

hbs2-peer/lib/HBS2/Peer/Notify.hs

@@ -91,4 +91,3 @@ instance ForRefLogEvents L4Proto => HasProtocol UNIX  (NotifyProto (RefLogEvents
 
 
 
-

hbs2-peer/lib/HBS2/Peer/RefChanNotifyLog.hs

@@ -0,0 +1,17 @@
+
+{-# LANGUAGE AllowAmbiguousTypes #-}
+{-# Language UndecidableInstances #-}
+module HBS2.Peer.RefChanNotifyLog where
+
+import HBS2.Data.Types.Refs
+import HBS2.Net.Proto.Definition()
+import HBS2.Net.Proto.RefChan
+
+
+type RefChanNotifyLogKey e = SomeRefKey (String, RefChanId e)
+
+
+makeRefChanNotifyLogKey :: RefChanId e -> RefChanNotifyLogKey e
+makeRefChanNotifyLogKey rc = SomeRefKey ("refchan-notify-log-key", rc)
+
+