wip

2023-03-10 08:50:23 +04:00 · 2023-03-10 08:50:23 +04:00 · 9bded3d3af
parent afa4bb8247
commit 9bded3d3af
5 changed files with 52 additions and 44 deletions
--- a/hbs2-core/lib/HBS2/Net/Proto/Definition.hs
+++ b/hbs2-core/lib/HBS2/Net/Proto/Definition.hs
@ -95,8 +95,8 @@ instance HasProtocol UDP (PeerExchange UDP) where
  decode = either (const Nothing) Just . deserialiseOrFail
  encode = serialise

-instance HasProtocol UDP (LRef UDP) where
-  type instance ProtocolId (LRef UDP) = 7
+instance HasProtocol UDP (LRefProto UDP) where
+  type instance ProtocolId (LRefProto UDP) = 7
  type instance Encoded UDP = ByteString
  decode = either (const Nothing) Just . deserialiseOrFail
  encode = serialise
@ -122,7 +122,7 @@ instance Expires (SessionKey UDP (PeerHandshake UDP)) where
 instance Expires (EventKey UDP (PeerAnnounce UDP)) where
  expiresIn _ = Nothing

-instance Expires (EventKey UDP (LRef UDP)) where
+instance Expires (EventKey UDP (LRefProto UDP)) where
  expiresIn _ = Nothing


--- a/hbs2-core/lib/HBS2/Net/Proto/RefLinear.hs
+++ b/hbs2-core/lib/HBS2/Net/Proto/RefLinear.hs
@ -20,42 +20,41 @@ import Lens.Micro.Platform
 import Type.Reflection (someTypeRep)


-newtype AnnLRefNonce = AnnLRefNonce Word64
-                         deriving newtype (Num,Enum,Real,Integral)
-                         deriving stock (Ord,Eq,Generic,Show)
-
-instance Serialise AnnLRefNonce
-
-
-data LRef e
+data LRefProto e
  = AnnLRef (Hash HbSync) (Signed SignaturePresent (MutableRef e 'LinearRef))
+  | LRefGetVal (Hash HbSync)
    deriving stock (Generic)

-instance Serialise (Signature e) => Serialise (LRef e)
+instance Serialise (Signature e) => Serialise (LRefProto e)

 data LRefI e m =
  LRefI
-  { getBlockI         :: GetBlockI HbSync m
+  { getBlockI           :: GetBlockI HbSync m
  , tryUpdateLinearRefI :: TryUpdateLinearRefI e HbSync m
+  , getLRefValI         :: GetLRefValI e HbSync m
+  , announceLRefValI    :: AnnounceLRefValI e HbSync m
  }

 type GetBlockI h m = Hash h -> m (Maybe ByteString)

 type TryUpdateLinearRefI e h m = Hash h -> Signed SignatureVerified (MutableRef e 'LinearRef) -> m Bool

+type GetLRefValI e h m = Hash h -> m (Maybe (Signed SignaturePresent (MutableRef e 'LinearRef)))
+
+type AnnounceLRefValI e h m = Hash h -> m ()
+
 refLinearProto :: forall e m  .
    ( MonadIO m
-    , Response e (LRef e) m
+    , Response e (LRefProto e) m
    , HasCredentials e m
    , Serialise (PubKey 'Sign e)
    , Signatures e
    )
    => LRefI e m
-    -> LRef e
+    -> LRefProto e
    -> m ()
 refLinearProto LRefI{..} = \case

-    -- Анонс ссылки (уведомление о новом состоянии без запроса)
    AnnLRef h (lref@LinearMutableRefSigned{}) -> do
        creds <- getCredentials @e

@ -65,22 +64,8 @@ refLinearProto LRefI{..} = \case

            lift $ forM_ (verifyLinearMutableRefSigned (refOwner g) lref) \vlref -> do
                r <- tryUpdateLinearRefI h vlref
-                when r do
-                    -- FIXME: В случае успеха разослать анонс на другие ноды
-                    pure ()
+                when r (announceLRefValI h)

-- data instance EventKey e (LRef e) =
--   AnnLRefInfoKey
--   deriving stock (Typeable, Eq,Generic)
-
-- data instance Event e (LRef e) =
--   AnnLRefEvent (Peer e) (AnnLRefInfo e) PeerNonce
--   deriving stock (Typeable)
-
-- instance Typeable (AnnLRefInfo e) => Hashable (EventKey e (LRef e)) where
--   hashWithSalt salt _ = hashWithSalt salt (someTypeRep p)
--     where
--       p = Proxy @(AnnLRefInfo e)
-
-- instance EventType ( Event e ( LRef e) ) where
--   isPersistent = True
+    LRefGetVal h -> void $ runMaybeT do
+        slref <- MaybeT (getLRefValI h)
+        lift $ response (AnnLRef @e h slref)
--- a/hbs2-peer/app/PeerMain.hs
+++ b/hbs2-peer/app/PeerMain.hs
@ -361,7 +361,6 @@ forKnownPeers m = do
    pd' <- find (KnownPeerKey p) id
    maybe1 pd' (pure ()) (m p)

-- FIXME: implement mkLRefAdapter
 mkLRefAdapter :: forall e st block m .
    ( m ~ PeerM e IO
    , Signatures e
@ -372,11 +371,24 @@ mkLRefAdapter :: forall e st block m .
    => m (LRefI e (CredentialsM e (ResponseM e m)))
 mkLRefAdapter = do
  st <- getStorage
-  pure $
-    LRefI
-    { getBlockI = liftIO . getBlock st
-    , tryUpdateLinearRefI = \h lvref -> liftIO $ tryUpdateLinearRef (st) h lvref
-    }
+
+  let
+
+    getBlockI = liftIO . getBlock st
+
+    tryUpdateLinearRefI h = liftIO . tryUpdateLinearRef st h
+
+    getLRefValI h = (liftIO . runMaybeT) do
+        refvalraw <- MaybeT $ (readLinkRaw st h) `orLogError` "error reading ref val"
+        MaybeT $ pure ((either (const Nothing) Just
+                      . deserialiseOrFail @(Signed SignaturePresent (MutableRef e 'LinearRef))) refvalraw)
+                      `orLogError` "can not parse channel ref"
+
+    announceLRefValI h = do
+        -- FIXME: implement announceLRefValI
+        pure ()
+
+  pure LRefI {..}

 runPeer :: forall e . e ~ UDP => PeerOpts -> IO ()
 runPeer opts = Exception.handle myException $ do
@ -484,8 +496,6 @@ runPeer opts = Exception.handle myException $ do
            runPeerM penv $ do
              adapter <- mkAdapter
              lrefAdapter <- mkLRefAdapter
-              -- lrefAdapter :: LRefI UDP (CredentialsM UDP (ResponseM UDP (PeerM UDP IO)))
-              --     <- undefined :: (PeerM UDP IO) (LRefI UDP (CredentialsM UDP (ResponseM UDP (PeerM UDP IO))))
              env <- ask

              pnonce <- peerNonce @e
@ -632,7 +642,7 @@ runPeer opts = Exception.handle myException $ do
                                           . deserialiseOrFail @(Signed SignaturePresent (MutableRef e 'LinearRef))) refvalraw)
                                      `orLogError` "can not parse channel ref"

-                                  let annlref :: LRef UDP
+                                  let annlref :: LRefProto UDP
                                      annlref = AnnLRef @e h slref

                                  lift do
@ -708,6 +718,9 @@ runPeer opts = Exception.handle myException $ do
  let annAction h = do
        liftIO $ atomically $ writeTQueue rpcQ (ANNOUNCE h)

+  let annLRefAction h = do
+        liftIO $ atomically $ writeTQueue rpcQ (ANNLREF h)
+
  let pingAction pa = do
        that <- thatPeer (Proxy @(RPC e))
        liftIO $ atomically $ writeTQueue rpcQ (PING pa (Just that))
@ -747,6 +760,7 @@ runPeer opts = Exception.handle myException $ do
  let arpc = RpcAdapter pokeAction
                        dontHandle
                        annAction
+                        annLRefAction
                        pingAction
                        dontHandle
                        fetchAction
@ -833,6 +847,8 @@ withRPC o cmd = do
                    case cmd of
                      RPCAnnounce{} -> pause @'Seconds 0.1 >> liftIO exitSuccess

+                      RPCAnnLRef{} -> pause @'Seconds 0.1 >> liftIO exitSuccess
+
                      RPCFetch{} -> pause @'Seconds 0.1 >> liftIO exitSuccess

                      RPCPing{} -> do
@ -865,9 +881,11 @@ withRPC o cmd = do
  void $ waitAnyCatchCancel [mrpc, prpc]

  where
-    adapter q pq = RpcAdapter dontHandle
+    adapter q pq = RpcAdapter
+                          dontHandle
                          (liftIO . atomically . writeTQueue pq)
                          (const $ liftIO exitSuccess)
+                          (const $ liftIO exitSuccess)
                          (const $ notice "ping?")
                          (liftIO . atomically . writeTQueue q)
                          dontHandle
@ -883,6 +901,7 @@ runRpcCommand opt = \case
  POKE -> withRPC opt RPCPoke
  PING s _ -> withRPC opt (RPCPing s)
  ANNOUNCE h -> withRPC opt (RPCAnnounce h)
+  ANNLREF h -> withRPC opt (RPCAnnLRef h)
  FETCH h  -> withRPC opt (RPCFetch h)
  PEERS -> withRPC opt RPCPeers
  SETLOG s -> withRPC opt (RPCLogLevel s)
--- a/hbs2-peer/app/RPC.hs
+++ b/hbs2-peer/app/RPC.hs
@ -28,6 +28,7 @@ data RPC e =
  | RPCPong (PeerAddr e)
  | RPCPokeAnswer (PubKey 'Sign e)
  | RPCAnnounce (Hash HbSync)
+  | RPCAnnLRef (Hash HbSync)
  | RPCFetch (Hash HbSync)
  | RPCPeers
  | RPCPeersAnswer (PeerAddr e) (PubKey 'Sign e)
@ -57,6 +58,7 @@ data RpcAdapter e m =
  { rpcOnPoke        :: RPC e -> m ()
  , rpcOnPokeAnswer  :: PubKey 'Sign e -> m ()
  , rpcOnAnnounce    :: Hash HbSync -> m ()
+  , rpcOnAnnLRef     :: Hash HbSync -> m ()
  , rpcOnPing        :: PeerAddr e -> m ()
  , rpcOnPong        :: PeerAddr e -> m ()
  , rpcOnFetch       :: Hash HbSync -> m ()
@ -106,6 +108,7 @@ rpcHandler adapter = \case
    p@RPCPoke{}        -> rpcOnPoke adapter p
    (RPCPokeAnswer k)  -> rpcOnPokeAnswer adapter k
    (RPCAnnounce h)    -> rpcOnAnnounce adapter h
+    (RPCAnnLRef h)     -> rpcOnAnnLRef adapter h
    (RPCPing pa)       -> rpcOnPing adapter pa
    (RPCPong pa)       -> rpcOnPong adapter pa
    (RPCFetch h)       -> rpcOnFetch adapter h
--- a/hbs2-peer/hbs2-peer.cabal
+++ b/hbs2-peer/hbs2-peer.cabal
@ -93,6 +93,7 @@ common shared-properties
                     , MultiParamTypeClasses
                     , OverloadedStrings
                     , QuasiQuotes
+                     , RecordWildCards
                     , ScopedTypeVariables
                     , StandaloneDeriving
                     , TupleSections