erebos 0.1.2 → 0.1.3

raw patch · 13 files changed

+923/−66 lines, 13 filesdep ~basedep ~template-haskellPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, template-haskell

API changes (from Hackage documentation)

Files

• CHANGELOG.md +7/−0
• README.md +1/−1
• erebos.cabal +20/−5
• main/Main.hs +94/−20
• main/Test.hs +97/−7
• main/Test/Service.hs +36/−0
• main/Version.hs +4/−0
• src/Erebos/Chatroom.hs +314/−0
• src/Erebos/Network.hs +51/−3
• src/Erebos/Network/Protocol.hs +285/−30
• src/Erebos/Set.hs +8/−0
• src/Erebos/State.hs +2/−0
• src/Erebos/Storage/Merge.hs +4/−0

@@ -1,5 +1,12 @@ # Revision history for erebos +## 0.1.3 -- 2024-05-05++* Enable/disable network services by command-line parameters+* Tab-completion of command name+* Implemented streams in network protocol+* Compatibility with GHC up to 9.8+ ## 0.1.2 -- 2024-02-20  * Compatibility with GHC up to 9.6

@@ -5,7 +5,7 @@ messaging service. Local identity is created on the first run. Protocol and services specification is being written at: -[http://erebosprotocol.net](http://erebosprotocol.net)+[https://erebosprotocol.net/spec](https://erebosprotocol.net/spec)  Erebos identity is based on locally stored cryptographic keys, all communication is end-to-end encrypted. Multiple devices can be attached to the

@@ -1,7 +1,7 @@ Cabal-Version:       2.2  Name:                erebos-Version:             0.1.2+Version:             0.1.3 Synopsis:            Decentralized messaging and synchronization Description:     Library and simple CLI interface implementing the Erebos identity@@ -18,7 +18,7 @@     See README for usage of the CLI tool. License:             BSD-3-Clause License-File:        LICENSE-Homepage:            http://erebosprotocol.net+Homepage:            https://erebosprotocol.net/erebos Author:              Roman Smrž <roman.smrz@seznam.cz> Maintainer:          roman.smrz@seznam.cz Category:            Network@@ -33,15 +33,28 @@ Flag ice     Description:    Enable peer discovery with ICE support using pjproject +Flag ci+    description:    Options for CI testing+    default: False+    manual: True+ source-repository head     type:       git     location:   git://erebosprotocol.net/erebos  common common-    ghc-options:         -Wall+    ghc-options:+        -Wall+        -fdefer-typed-holes +    if flag(ci)+        ghc-options:+            -Werror+            -- sometimes needed for backward/forward compatibility:+            -Wno-error=unused-imports+     build-depends:-        base >=4.13 && <4.19,+        base >=4.13 && <4.20,      default-extensions:         DefaultSignatures@@ -82,6 +95,7 @@     exposed-modules:         Erebos.Attach         Erebos.Channel+        Erebos.Chatroom         Erebos.Contact         Erebos.Identity         Erebos.Message@@ -156,6 +170,7 @@     other-modules:         Paths_erebos         Test+        Test.Service         Version         Version.Git     autogen-modules:@@ -170,7 +185,7 @@         mtl,         network,         process >=1.6 && <1.7,-        template-haskell >=2.18 && <2.22,+        template-haskell >=2.17 && <2.22,         text,         time,         transformers >= 0.5 && <0.7,

@@ -5,6 +5,7 @@  import Control.Arrow (first) import Control.Concurrent+import Control.Exception import Control.Monad import Control.Monad.Except import Control.Monad.Reader@@ -30,6 +31,8 @@ import System.Console.GetOpt import System.Console.Haskeline import System.Environment+import System.Exit+import System.IO  import Erebos.Attach import Erebos.Contact@@ -53,29 +56,74 @@  data Options = Options     { optServer :: ServerOptions+    , optServices :: [ServiceOption]+    , optShowHelp :: Bool     , optShowVersion :: Bool     } +data ServiceOption = ServiceOption+    { soptName :: String+    , soptService :: SomeService+    , soptEnabled :: Bool+    , soptDescription :: String+    }+ defaultOptions :: Options defaultOptions = Options     { optServer = defaultServerOptions+    , optServices = availableServices+    , optShowHelp = False     , optShowVersion = False     } +availableServices :: [ServiceOption]+availableServices =+    [ ServiceOption "attach" (someService @AttachService Proxy)+        True "attach (to) other devices"+    , ServiceOption "sync" (someService @SyncService Proxy)+        True "synchronization with attached devices"+    , ServiceOption "contact" (someService @ContactService Proxy)+        True "create contacts with network peers"+    , ServiceOption "dm" (someService @DirectMessage Proxy)+        True "direct messages"+#ifdef ENABLE_ICE_SUPPORT+    , ServiceOption "discovery" (someService @DiscoveryService Proxy)+        True "peer discovery"+#endif+    ]+ options :: [OptDescr (Options -> Options)] options =     [ Option ['p'] ["port"]-        (ReqArg (\p -> so $ \opts -> opts { serverPort = read p }) "PORT")+        (ReqArg (\p -> so $ \opts -> opts { serverPort = read p }) "<port>")         "local port to bind"     , Option ['s'] ["silent"]         (NoArg (so $ \opts -> opts { serverLocalDiscovery = False }))         "do not send announce packets for local discovery"+    , Option ['h'] ["help"]+        (NoArg $ \opts -> opts { optShowHelp = True })+        "show this help and exit"     , Option ['V'] ["version"]         (NoArg $ \opts -> opts { optShowVersion = True })         "show version and exit"     ]     where so f opts = opts { optServer = f $ optServer opts } +servicesOptions :: [OptDescr (Options -> Options)]+servicesOptions = concatMap helper $ "all" : map soptName availableServices+  where+    helper name =+        [ Option [] ["enable-" <> name] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = True }) ""+        , Option [] ["disable-" <> name] (NoArg $ so $ change name $ \sopt -> sopt { soptEnabled = False }) ""+        ]+    so f opts = opts { optServices = f $ optServices opts }+    change :: String -> (ServiceOption -> ServiceOption) -> [ServiceOption] -> [ServiceOption]+    change name f (s : ss)+        | soptName s == name || name == "all"+                    = f s : change name f ss+        | otherwise =   s : change name f ss+    change _ _ []   = []+ main :: IO () main = do     st <- liftIO $ openStorage . fromMaybe "./.erebos" =<< lookupEnv "EREBOS_DIR"@@ -129,19 +177,39 @@          ["test"] -> runTestTool st -        args -> do-            opts <- case getOpt Permute options args of-                (o, [], []) -> return (foldl (flip id) defaultOptions o)-                (_, _, errs) -> ioError (userError (concat errs ++ usageInfo header options))-                    where header = "Usage: erebos [OPTION...]"+        args -> case getOpt Permute (options ++ servicesOptions) args of+            (o, [], []) -> do+                let opts = foldl (flip id) defaultOptions o+                    header = "Usage: erebos [OPTION...]"+                    serviceDesc ServiceOption {..} = padService ("  " <> soptName) <> soptDescription -            if optShowVersion opts-               then putStrLn versionLine-               else interactiveLoop st opts+                    padTo n str = str <> replicate (n - length str) ' '+                    padOpt = padTo 28+                    padService = padTo 16 +                if | optShowHelp opts -> putStr $ usageInfo header options <> unlines+                      (+                        [ padOpt "  --enable-<service>"  <> "enable network service <service>"+                        , padOpt "  --disable-<service>" <> "disable network service <service>"+                        , padOpt "  --enable-all"        <> "enable all network services"+                        , padOpt "  --disable-all"       <> "disable all network services"+                        , ""+                        , "Available network services:"+                        ] ++ map serviceDesc availableServices+                      )+                   | optShowVersion opts -> putStrLn versionLine+                   | otherwise -> interactiveLoop st opts+            (_, _, errs) -> do+                progName <- getProgName+                hPutStrLn stderr $ concat errs <> "Try `" <> progName <> " --help' for more information."+                exitFailure ++inputSettings :: Settings IO+inputSettings = setComplete commandCompletion $ defaultSettings+ interactiveLoop :: Storage -> Options -> IO ()-interactiveLoop st opts = runInputT defaultSettings $ do+interactiveLoop st opts = runInputT inputSettings $ do     erebosHead <- liftIO $ loadLocalStateHead st     outputStrLn $ T.unpack $ displayIdentity $ headLocalIdentity erebosHead @@ -157,15 +225,8 @@             extPrintLn . formatMessage tzone . fromStored      server <- liftIO $ do-        startServer (optServer opts) erebosHead extPrintLn-            [ someService @AttachService Proxy-            , someService @SyncService Proxy-            , someService @ContactService Proxy-            , someService @DirectMessage Proxy-#ifdef ENABLE_ICE_SUPPORT-            , someService @DiscoveryService Proxy-#endif-            ]+        startServer (optServer opts) erebosHead extPrintLn $+            map soptService $ filter soptEnabled $ optServices opts      peers <- liftIO $ newMVar []     contextOptions <- liftIO $ newMVar []@@ -262,11 +323,16 @@                     | SelectedContact Contact  newtype CommandM a = CommandM (ReaderT CommandInput (StateT CommandState (ExceptT String IO)) a)-    deriving (Functor, Applicative, Monad, MonadIO, MonadReader CommandInput, MonadState CommandState, MonadError String)+    deriving (Functor, Applicative, Monad, MonadReader CommandInput, MonadState CommandState, MonadError String)  instance MonadFail CommandM where     fail = throwError +instance MonadIO CommandM where+    liftIO act = CommandM (liftIO (try act)) >>= \case+        Left (e :: SomeException) -> throwError (show e)+        Right x -> return x+ instance MonadRandom CommandM where     getRandomBytes = liftIO . getRandomBytes @@ -321,6 +387,14 @@     , ("ice-send", cmdIceSend) #endif     ]++commandCompletion :: CompletionFunc IO+commandCompletion = completeWordWithPrev Nothing [ ' ', '\t', '\n', '\r' ] $ curry $ \case+    ([], '/':pref) -> return . map (simpleCompletion . ('/':)) . filter (pref `isPrefixOf`) $ sortedCommandNames+    _ -> return []+  where+    sortedCommandNames = sort $ map fst commands+  cmdUnknown :: String -> Command cmdUnknown cmd = liftIO $ putStrLn $ "Unknown command: " ++ cmd

@@ -29,6 +29,7 @@ import System.IO.Error  import Erebos.Attach+import Erebos.Chatroom import Erebos.Contact import Erebos.Identity import Erebos.Message@@ -43,7 +44,9 @@ import Erebos.Storage.Merge import Erebos.Sync +import Test.Service + data TestState = TestState     { tsHead :: Maybe (Head LocalState)     , tsServer :: Maybe RunningServer@@ -243,6 +246,7 @@     , ("start-server", cmdStartServer)     , ("stop-server", cmdStopServer)     , ("peer-add", cmdPeerAdd)+    , ("test-message-send", cmdTestMessageSend)     , ("shared-state-get", cmdSharedStateGet)     , ("shared-state-wait", cmdSharedStateWait)     , ("watch-local-identity", cmdWatchLocalIdentity)@@ -261,6 +265,10 @@     , ("dm-send-contact", cmdDmSendContact)     , ("dm-list-peer", cmdDmListPeer)     , ("dm-list-contact", cmdDmListContact)+    , ("chatroom-create", cmdChatroomCreate)+    , ("chatroom-list-local", cmdChatroomListLocal)+    , ("chatroom-watch-local", cmdChatroomWatchLocal)+    , ("chatroom-set-name", cmdChatroomSetName)     ]  cmdStore :: Command@@ -307,6 +315,26 @@         cmdOut $ "stored-set-item" ++ concatMap ((' ':) . show . refDigest . storedRef) item     cmdOut $ "stored-set-done" +initTestHead :: Head LocalState -> Command+initTestHead h = do+    _ <- liftIO . watchReceivedMessages h . dmReceivedWatcher =<< asks tiOutput+    modify $ \s -> s { tsHead = Just h }++loadTestHead :: CommandM (Head LocalState)+loadTestHead = do+    st <- asks tiStorage+    h <- loadHeads st >>= \case+        h : _ -> return h+        [] -> fail "no local head found"+    initTestHead h+    return h++getOrLoadHead :: CommandM (Head LocalState)+getOrLoadHead = do+    gets tsHead >>= \case+        Just h -> return h+        Nothing -> loadTestHead+ cmdCreateIdentity :: Command cmdCreateIdentity = do     st <- asks tiStorage@@ -325,21 +353,24 @@             { lsIdentity = idExtData identity             , lsShared = shared             }--    _ <- liftIO . watchReceivedMessages h . dmReceivedWatcher =<< asks tiOutput-    modify $ \s -> s { tsHead = Just h }+    initTestHead h  cmdStartServer :: Command cmdStartServer = do     out <- asks tiOutput -    Just h <- gets tsHead+    h <- getOrLoadHead     rsPeers <- liftIO $ newMVar (1, [])     rsServer <- liftIO $ startServer defaultServerOptions h (hPutStrLn stderr)         [ someServiceAttr $ pairingAttributes (Proxy @AttachService) out rsPeers "attach"         , someServiceAttr $ pairingAttributes (Proxy @ContactService) out rsPeers "contact"         , someServiceAttr $ directMessageAttributes out         , someService @SyncService Proxy+        , someService @ChatroomService Proxy+        , someServiceAttr $ (defaultServiceAttributes Proxy)+            { testMessageReceived = \otype len sref ->+                liftIO $ outLine out $ unwords ["test-message-received", otype, len, sref]+            }         ]      rsPeerThread <- liftIO $ forkIO $ void $ forever $ do@@ -378,6 +409,15 @@     addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just host) (Just port)     void $ liftIO $ serverPeer rsServer (addrAddress addr) +cmdTestMessageSend :: Command+cmdTestMessageSend = do+    [spidx, tref] <- asks tiParams+    st <- asks tiStorage+    Just ref <- liftIO $ readRef st (encodeUtf8 tref)+    peer <- getPeer spidx+    sendToPeer peer $ TestMessage $ wrappedLoad ref+    cmdOut "test-message-send done"+ cmdSharedStateGet :: Command cmdSharedStateGet = do     h <- getHead@@ -387,7 +427,7 @@ cmdSharedStateWait = do     st <- asks tiStorage     out <- asks tiOutput-    Just h <- gets tsHead+    h <- getOrLoadHead     trefs <- asks tiParams      liftIO $ do@@ -403,7 +443,7 @@  cmdWatchLocalIdentity :: Command cmdWatchLocalIdentity = do-    Just h <- gets tsHead+    h <- getOrLoadHead     Nothing <- gets tsWatchedLocalIdentity      out <- asks tiOutput@@ -413,7 +453,7 @@  cmdWatchSharedIdentity :: Command cmdWatchSharedIdentity = do-    Just h <- gets tsHead+    h <- getOrLoadHead     Nothing <- gets tsWatchedSharedIdentity      out <- asks tiOutput@@ -549,3 +589,53 @@     [cid] <- asks tiParams     Just to <- contactIdentity <$> getContact cid     dmList to++cmdChatroomCreate :: Command+cmdChatroomCreate = do+    [name] <- asks tiParams+    room <- createChatroom (Just name) Nothing+    cmdOut $ unwords $ "chatroom-create-done" : chatroomInfo room++getChatroomStateData :: Text -> CommandM (Stored ChatroomStateData)+getChatroomStateData tref = do+    st <- asks tiStorage+    Just ref <- liftIO $ readRef st (encodeUtf8 tref)+    return $ wrappedLoad ref++cmdChatroomSetName :: Command+cmdChatroomSetName = do+    [cid, name] <- asks tiParams+    sdata <- getChatroomStateData cid+    updateChatroomByStateData sdata (Just name) Nothing >>= \case+        Just room -> cmdOut $ unwords $ "chatroom-set-name-done" : chatroomInfo room+        Nothing -> cmdOut "chatroom-set-name-failed"++cmdChatroomListLocal :: Command+cmdChatroomListLocal = do+    [] <- asks tiParams+    rooms <- listChatrooms+    forM_ rooms $ \room -> do+        cmdOut $ unwords $ "chatroom-list-item" : chatroomInfo room+    cmdOut "chatroom-list-done"++cmdChatroomWatchLocal :: Command+cmdChatroomWatchLocal = do+    [] <- asks tiParams+    h <- getHead+    out <- asks tiOutput+    void $ watchChatrooms h $ \_ -> \case+        Nothing -> return ()+        Just diff -> forM_ diff $ \case+            AddedChatroom room -> outLine out $ unwords $ "chatroom-watched-added" : chatroomInfo room+            RemovedChatroom room -> outLine out $ unwords $ "chatroom-watched-removed" : chatroomInfo room+            UpdatedChatroom oldroom room -> outLine out $ unwords $ concat+                [ [ "chatroom-watched-updated" ], chatroomInfo room+                , [ "old" ], map (show . refDigest . storedRef) (roomStateData oldroom)+                , [ "new" ], map (show . refDigest . storedRef) (roomStateData room)+                ]++chatroomInfo :: ChatroomState -> [String]+chatroomInfo room =+    [ show . refDigest . storedRef . head . filterAncestors . concatMap storedRoots . toComponents $ room+    , maybe "<unnamed>" T.unpack $ roomName =<< roomStateRoom room+    ]

@@ -0,0 +1,36 @@+module Test.Service (+    TestMessage(..),+    TestMessageAttributes(..),+) where++import Control.Monad.Reader++import Data.ByteString.Lazy.Char8 qualified as BL++import Erebos.Network+import Erebos.Service+import Erebos.Storage++data TestMessage = TestMessage (Stored Object)++data TestMessageAttributes = TestMessageAttributes+    { testMessageReceived :: String -> String -> String -> ServiceHandler TestMessage ()+    }++instance Storable TestMessage where+    store' (TestMessage msg) = store' msg+    load' = TestMessage <$> load'++instance Service TestMessage where+    serviceID _ = mkServiceID "cb46b92c-9203-4694-8370-8742d8ac9dc8"++    type ServiceAttributes TestMessage = TestMessageAttributes+    defaultServiceAttributes _ = TestMessageAttributes (\_ _ _ -> return ())++    serviceHandler smsg = do+        let TestMessage sobj = fromStored smsg+        case map BL.unpack $ BL.words $ BL.takeWhile (/='\n') $ serializeObject $ fromStored sobj of+            [otype, len] -> do+                cb <- asks $ testMessageReceived . svcAttributes+                cb otype len (show $ refDigest $ storedRef sobj)+            _ -> return ()

@@ -1,5 +1,9 @@ {-# LANGUAGE TemplateHaskell #-} +-- "Pattern match is redundant" warning can be generated based on template+-- haskell $$tGitVersion value+{-# OPTIONS_GHC -Wno-error=overlapping-patterns #-}+ module Version (     versionLine, ) where

@@ -0,0 +1,314 @@+module Erebos.Chatroom (+    Chatroom(..),+    ChatroomData(..),+    validateChatroom,++    ChatroomState(..),+    ChatroomStateData(..),+    createChatroom,+    updateChatroomByStateData,+    listChatrooms,+    findChatroomByRoomData,+    findChatroomByStateData,++    ChatroomSetChange(..),+    watchChatrooms,++    ChatroomService(..),+) where++import Control.Arrow+import Control.Monad+import Control.Monad.Except+import Control.Monad.IO.Class++import Data.IORef+import Data.List+import Data.Maybe+import Data.Monoid+import Data.Ord+import Data.Text (Text)++import Erebos.PubKey+import Erebos.Service+import Erebos.Set+import Erebos.State+import Erebos.Storage+import Erebos.Storage.Merge+import Erebos.Util+++data ChatroomData = ChatroomData+    { rdPrev :: [Stored (Signed ChatroomData)]+    , rdName :: Maybe Text+    , rdDescription :: Maybe Text+    , rdKey :: Stored PublicKey+    }++data Chatroom = Chatroom+    { roomData :: [Stored (Signed ChatroomData)]+    , roomName :: Maybe Text+    , roomDescription :: Maybe Text+    , roomKey :: Stored PublicKey+    }++instance Storable ChatroomData where+    store' ChatroomData {..} = storeRec $ do+        mapM_ (storeRef "SPREV") rdPrev+        storeMbText "name" rdName+        storeMbText "description" rdDescription+        storeRef "key" rdKey++    load' = loadRec $ do+        rdPrev <- loadRefs "SPREV"+        rdName <- loadMbText "name"+        rdDescription <- loadMbText "description"+        rdKey <- loadRef "key"+        return ChatroomData {..}++validateChatroom :: [Stored (Signed ChatroomData)] -> Except String Chatroom+validateChatroom roomData = do+    when (null roomData) $ throwError "null data"+    when (not $ getAll $ walkAncestors verifySignatures roomData) $ do+        throwError "signature verification failed"++    let roomName = findPropertyFirst (rdName . fromStored . signedData) roomData+        roomDescription = findPropertyFirst (rdDescription . fromStored . signedData) roomData+    roomKey <- maybe (throwError "missing key") return $+        findPropertyFirst (Just . rdKey . fromStored . signedData) roomData+    return Chatroom {..}+  where+    verifySignatures sdata =+        let rdata = fromSigned sdata+            required = concat+                [ [ rdKey rdata ]+                , map (rdKey . fromSigned) $ rdPrev rdata+                ]+         in All $ all (fromStored sdata `isSignedBy`) required+++data ChatroomStateData = ChatroomStateData+    { rsdPrev :: [Stored ChatroomStateData]+    , rsdRoom :: [Stored (Signed ChatroomData)]+    }++data ChatroomState = ChatroomState+    { roomStateData :: [Stored ChatroomStateData]+    , roomStateRoom :: Maybe Chatroom+    }++instance Storable ChatroomStateData where+    store' ChatroomStateData {..} = storeRec $ do+        forM_ rsdPrev $ storeRef "PREV"+        forM_ rsdRoom $ storeRef "room"++    load' = loadRec $ do+        rsdPrev <- loadRefs "PREV"+        rsdRoom <- loadRefs "room"+        return ChatroomStateData {..}++instance Mergeable ChatroomState where+    type Component ChatroomState = ChatroomStateData++    mergeSorted cdata = ChatroomState+        { roomStateData = cdata+        , roomStateRoom = either (const Nothing) Just $ runExcept $+            validateChatroom $ concat $ findProperty ((\case [] -> Nothing; xs -> Just xs) . rsdRoom) cdata+        }++    toComponents = roomStateData++instance SharedType (Set ChatroomState) where+    sharedTypeID _ = mkSharedTypeID "7bc71cbf-bc43-42b1-b413-d3a2c9a2aae0"++createChatroom :: (MonadStorage m, MonadHead LocalState m, MonadIO m, MonadError String m) => Maybe Text -> Maybe Text -> m ChatroomState+createChatroom rdName rdDescription = do+    (secret, rdKey) <- liftIO . generateKeys =<< getStorage+    let rdPrev = []+    rdata <- mstore =<< sign secret =<< mstore ChatroomData {..}+    cstate <- mergeSorted . (:[]) <$> mstore ChatroomStateData+        { rsdPrev = []+        , rsdRoom = [ rdata ]+        }++    updateLocalHead $ updateSharedState $ \rooms -> do+        st <- getStorage+        (, cstate) <$> storeSetAdd st cstate rooms++findAndUpdateChatroomState+    :: (MonadStorage m, MonadHead LocalState m)+    => (ChatroomState -> Maybe (m ChatroomState))+    -> m (Maybe ChatroomState)+findAndUpdateChatroomState f = do+    updateLocalHead $ updateSharedState $ \roomSet -> do+        let roomList = fromSetBy (comparing $ roomName <=< roomStateRoom) roomSet+        case catMaybes $ map (\x -> (x,) <$> f x) roomList of+            ((orig, act) : _) -> do+                upd <- act+                if roomStateData orig /= roomStateData upd+                  then do+                    st <- getStorage+                    roomSet' <- storeSetAdd st upd roomSet+                    return (roomSet', Just upd)+                  else do+                    return (roomSet, Just upd)+            [] -> return (roomSet, Nothing)++updateChatroomByStateData+    :: (MonadStorage m, MonadHead LocalState m, MonadError String m)+    => Stored ChatroomStateData+    -> Maybe Text+    -> Maybe Text+    -> m (Maybe ChatroomState)+updateChatroomByStateData lookupData newName newDesc = findAndUpdateChatroomState $ \cstate -> do+    guard $ any (lookupData `precedesOrEquals`) $ roomStateData cstate+    room <- roomStateRoom cstate+    Just $ do+        secret <- loadKey $ roomKey room+        rdata <- mstore =<< sign secret =<< mstore ChatroomData+            { rdPrev = roomData room+            , rdName = newName+            , rdDescription = newDesc+            , rdKey = roomKey room+            }+        mergeSorted . (:[]) <$> mstore ChatroomStateData+            { rsdPrev = roomStateData cstate+            , rsdRoom = [ rdata ]+            }+++listChatrooms :: MonadHead LocalState m => m [ChatroomState]+listChatrooms = fromSetBy (comparing $ roomName <=< roomStateRoom) .+    lookupSharedValue . lsShared . fromStored <$> getLocalHead++findChatroom :: MonadHead LocalState m => (ChatroomState -> Bool) -> m (Maybe ChatroomState)+findChatroom p = do+    list <- map snd . chatroomSetToList . lookupSharedValue . lsShared . fromStored <$> getLocalHead+    return $ find p list++findChatroomByRoomData :: MonadHead LocalState m => Stored (Signed ChatroomData) -> m (Maybe ChatroomState)+findChatroomByRoomData cdata = findChatroom $+    maybe False (any (cdata `precedesOrEquals`) . roomData) . roomStateRoom++findChatroomByStateData :: MonadHead LocalState m => Stored ChatroomStateData -> m (Maybe ChatroomState)+findChatroomByStateData cdata = findChatroom $ any (cdata `precedesOrEquals`) . roomStateData+++data ChatroomSetChange = AddedChatroom ChatroomState+                       | RemovedChatroom ChatroomState+                       | UpdatedChatroom ChatroomState ChatroomState++watchChatrooms :: MonadIO m => Head LocalState -> (Set ChatroomState -> Maybe [ChatroomSetChange] -> IO ()) -> m WatchedHead+watchChatrooms h f = liftIO $ do+    lastVar <- newIORef Nothing+    watchHeadWith h (lookupSharedValue . lsShared . headObject) $ \cur -> do+        let curList = chatroomSetToList cur+        mbLast <- readIORef lastVar+        writeIORef lastVar $ Just curList+        f cur $ do+            lastList <- mbLast+            return $ makeChatroomDiff lastList curList++chatroomSetToList :: Set ChatroomState -> [(Stored ChatroomStateData, ChatroomState)]+chatroomSetToList = map (cmp &&& id) . fromSetBy (comparing cmp)+  where+    cmp :: ChatroomState -> Stored ChatroomStateData+    cmp = head . filterAncestors . concatMap storedRoots . toComponents++makeChatroomDiff+    :: [(Stored ChatroomStateData, ChatroomState)]+    -> [(Stored ChatroomStateData, ChatroomState)]+    -> [ChatroomSetChange]+makeChatroomDiff (x@(cx, vx) : xs) (y@(cy, vy) : ys)+    | cx < cy = RemovedChatroom vx : makeChatroomDiff xs (y : ys)+    | cx > cy = AddedChatroom vy : makeChatroomDiff (x : xs) ys+    | roomStateData vx /= roomStateData vy = UpdatedChatroom vx vy : makeChatroomDiff xs ys+    | otherwise = makeChatroomDiff xs ys+makeChatroomDiff xs [] = map (RemovedChatroom . snd) xs+makeChatroomDiff [] ys = map (AddedChatroom . snd) ys+++data ChatroomService = ChatroomService+    { chatRoomQuery :: Bool+    , chatRoomInfo :: [Stored (Signed ChatroomData)]+    }++emptyPacket :: ChatroomService+emptyPacket = ChatroomService+    { chatRoomQuery = False+    , chatRoomInfo = []+    }++instance Storable ChatroomService where+    store' ChatroomService {..} = storeRec $ do+        when  chatRoomQuery $ storeEmpty "room-query"+        forM_ chatRoomInfo $ storeRef "room-info"++    load' = loadRec $ do+        chatRoomQuery <- isJust <$> loadMbEmpty "room-query"+        chatRoomInfo <- loadRefs "room-info"+        return ChatroomService {..}++data PeerState = PeerState+    { psSendRoomUpdates :: Bool+    , psLastList :: [(Stored ChatroomStateData, ChatroomState)]+    }++instance Service ChatroomService where+    serviceID _ = mkServiceID "627657ae-3e39-468a-8381-353395ef4386"++    type ServiceState ChatroomService = PeerState+    emptyServiceState _ = PeerState+        { psSendRoomUpdates = False+        , psLastList = []+        }++    serviceHandler spacket = do+        let ChatroomService {..} = fromStored spacket+        svcModify $ \s -> s { psSendRoomUpdates = True }++        when chatRoomQuery $ do+            rooms <- listChatrooms+            replyPacket emptyPacket+                { chatRoomInfo = concatMap roomData $ catMaybes $ map roomStateRoom rooms+                }++        when (not $ null chatRoomInfo) $ do+            updateLocalHead_ $ updateSharedState_ $ \roomSet -> do+                let rooms = fromSetBy (comparing $ roomName <=< roomStateRoom) roomSet+                    upd set (roomInfo :: Stored (Signed ChatroomData)) = do+                        let currentRoots = storedRoots roomInfo+                            isCurrentRoom = any ((`intersectsSorted` currentRoots) . storedRoots) .+                                concatMap (rsdRoom . fromStored) . roomStateData++                        let prev = concatMap roomStateData $ filter isCurrentRoom rooms+                            prevRoom = concatMap (rsdRoom . fromStored) prev+                            room = filterAncestors $ (roomInfo : ) prevRoom++                        -- update local state only if we got roomInfo not present there+                        if roomInfo `notElem` prevRoom && roomInfo `elem` room+                          then do+                            sdata <- mstore ChatroomStateData { rsdPrev = prev, rsdRoom = room }+                            storeSetAddComponent sdata set+                          else return set+                foldM upd roomSet chatRoomInfo++    serviceNewPeer = do+        replyPacket emptyPacket { chatRoomQuery = True }++    serviceStorageWatchers _ = (:[]) $+        SomeStorageWatcher (lookupSharedValue . lsShared . fromStored) syncChatroomsToPeer++syncChatroomsToPeer :: Set ChatroomState -> ServiceHandler ChatroomService ()+syncChatroomsToPeer set = do+    ps@PeerState {..} <- svcGet+    when psSendRoomUpdates $ do+        let curList = chatroomSetToList set+        updates <- fmap (concat . catMaybes) $+            forM (makeChatroomDiff psLastList curList) $ return . \case+                AddedChatroom room -> roomData <$> roomStateRoom room+                RemovedChatroom {} -> Nothing+                UpdatedChatroom _ room -> roomData <$> roomStateRoom room+        when (not $ null updates) $ do+            replyPacket $ emptyPacket { chatRoomInfo = updates }+        svcSet $ ps { psLastList = curList }

@@ -30,7 +30,8 @@ import Control.Monad.Reader import Control.Monad.State -import qualified Data.ByteString.Char8 as BC+import Data.ByteString.Char8 qualified as BC+import Data.ByteString.Lazy qualified as BL import Data.Function import Data.IP qualified as IP import Data.List@@ -179,7 +180,12 @@ lookupServiceType (_ : hs) = lookupServiceType hs lookupServiceType [] = Nothing +lookupNewStreams :: [TransportHeaderItem] -> [Word8]+lookupNewStreams (StreamOpen num : rest) = num : lookupNewStreams rest+lookupNewStreams (_ : rest) = lookupNewStreams rest+lookupNewStreams [] = [] + newWaitingRef :: RefDigest -> (Ref -> WaitingRefCallback) -> PacketHandler WaitingRef newWaitingRef dgst act = do     peer@Peer {..} <- gets phPeer@@ -421,6 +427,28 @@ addBody :: Ref -> PacketHandler () addBody r = modify $ \ph -> ph { phBody = r `appendDistinct` phBody ph } +openStream :: PacketHandler RawStreamWriter+openStream = do+    Peer {..} <- gets phPeer+    conn <- readTVarP peerConnection >>= \case+        Right conn -> return conn+        _          -> throwError "can't open stream without established connection"+    (hdr, writer, handler) <- liftSTM (connAddWriteStream conn) >>= \case+        Right res -> return res+        Left err -> throwError err++    liftSTM $ writeTQueue (serverIOActions peerServer_) (liftIO $ forkServerThread peerServer_ handler)+    addHeader hdr+    return writer++acceptStream :: Word8 -> PacketHandler RawStreamReader+acceptStream streamNumber = do+    Peer {..} <- gets phPeer+    conn <- readTVarP peerConnection >>= \case+        Right conn -> return conn+        _          -> throwError "can't accept stream without established connection"+    liftSTM $ connAddReadStream conn streamNumber+ appendDistinct :: Eq a => a -> [a] -> [a] appendDistinct x (y:ys) | x == y    = y : ys                         | otherwise = y : appendDistinct x ys@@ -461,7 +489,15 @@                         partialRefFromDigest (peerInStorage peer) dgst                     addHeader $ DataResponse dgst                     addAckedBy [ Acknowledged dgst, Rejected dgst ]-                    addBody $ mref+                    let bytes = lazyLoadBytes mref+                    -- TODO: MTU+                    if (secure && BL.length bytes > 500)+                      then do+                        stream <- openStream+                        liftSTM $ writeTQueue (serverIOActions server) $ void $ liftIO $ forkIO $ do+                            writeByteStringToStream stream bytes+                      else do+                        addBody $ mref                 | otherwise -> do                     logd $ "unauthorized data request for " ++ show dgst                     addHeader $ Rejected dgst@@ -471,6 +507,18 @@                     when (not secure) $ do                         addHeader $ Acknowledged dgst                     liftSTM $ writeTQueue (serverDataResponse server) (peer, Just pref)++                | streamNumber : _ <- lookupNewStreams headers -> do+                    streamReader <- acceptStream streamNumber+                    liftSTM $ writeTQueue (serverIOActions server) $ void $ liftIO $ forkIO $ do+                        (runExcept <$> readObjectsFromStream (peerInStorage peer) streamReader) >>= \case+                            Left err -> atomically $ writeTQueue (serverErrorLog server) $+                                "failed to receive object from stream: " <> err+                            Right objs -> do+                                forM_ objs $ \obj -> do+                                    pref <- storeObject (peerInStorage peer) obj+                                    atomically $ writeTQueue (serverDataResponse server) (peer, Just pref)+                 | otherwise -> throwError $ "mismatched data response " ++ show dgst              AnnounceSelf dgst@@ -708,7 +756,7 @@         ServiceReply (Right sx) use -> return $ Just (storedRef sx, use)         ServiceFinally act -> act >> return Nothing     let dgsts = map (refDigest . fst) srefs-    let content = map fst $ filter snd srefs+    let content = map fst $ filter (\(ref, use) -> use && BL.length (lazyLoadBytes ref) < 500) srefs -- TODO: MTU         header = TransportHeader (ServiceType (serviceID $ head parts) : map ServiceRef dgsts)         packet = TransportPacket header content         ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- dgsts ]

@@ -20,6 +20,13 @@     connGetChannel,     connSetChannel, +    RawStreamReader, RawStreamWriter,+    connAddWriteStream,+    connAddReadStream,+    readStreamToList,+    readObjectsFromStream,+    writeByteStringToStream,+     module Erebos.Flow, ) where @@ -29,6 +36,7 @@ import Control.Concurrent.STM import Control.Monad import Control.Monad.Except+import Control.Monad.Trans  import Data.Bits import Data.ByteString (ByteString)@@ -39,6 +47,8 @@ import Data.Maybe import Data.Text (Text) import Data.Text qualified as T+import Data.Void+import Data.Word  import System.Clock @@ -77,6 +87,7 @@     | TrChannelAccept RefDigest     | ServiceType ServiceID     | ServiceRef RefDigest+    | StreamOpen Word8     deriving (Eq, Show)  newtype Cookie = Cookie ByteString@@ -111,6 +122,7 @@               TrChannelAccept dgst -> (BC.pack "CAC", RecRef $ partialRefFromDigest st dgst)               ServiceType stype -> (BC.pack "SVT", RecUUID $ toUUID stype)               ServiceRef dgst -> (BC.pack "SVR", RecRef $ partialRefFromDigest st dgst)+              StreamOpen num -> (BC.pack "STO", RecInt $ fromIntegral num)  transportFromObject :: PartialObject -> Maybe TransportHeader transportFromObject (Rec items) = case catMaybes $ map single items of@@ -132,6 +144,7 @@               | name == BC.pack "CAC", RecRef ref <- content -> Just $ TrChannelAccept $ refDigest ref               | name == BC.pack "SVT", RecUUID uuid <- content -> Just $ ServiceType $ fromUUID uuid               | name == BC.pack "SVR", RecRef ref <- content -> Just $ ServiceRef $ refDigest ref+              | name == BC.pack "STO", RecInt num <- content -> Just $ StreamOpen $ fromIntegral num               | otherwise -> Nothing transportFromObject _ = Nothing @@ -150,13 +163,18 @@     }  data Connection addr = Connection-    { cAddress :: addr+    { cGlobalState :: GlobalState addr+    , cAddress :: addr     , cDataUp :: Flow (Bool, TransportPacket PartialObject) (Bool, TransportPacket Ref, [TransportHeaderItem])     , cDataInternal :: Flow (Bool, TransportPacket Ref, [TransportHeaderItem]) (Bool, TransportPacket PartialObject)     , cChannel :: TVar ChannelState     , cSecureOutQueue :: TQueue (Bool, TransportPacket Ref, [TransportHeaderItem])+    , cMaxInFlightPackets :: TVar Int+    , cReservedPackets :: TVar Int     , cSentPackets :: TVar [SentPacket]     , cToAcknowledge :: TVar [Integer]+    , cInStreams :: TVar [(Word8, Stream)]+    , cOutStreams :: TVar [(Word8, Stream)]     }  connAddress :: Connection addr -> addr@@ -172,7 +190,168 @@ connSetChannel Connection {..} ch = do     writeTVar cChannel ch +connAddWriteStream :: Connection addr -> STM (Either String (TransportHeaderItem, RawStreamWriter, IO ()))+connAddWriteStream conn@Connection {..} = do+    outStreams <- readTVar cOutStreams+    let doInsert :: Word8 -> [(Word8, Stream)] -> ExceptT String STM ((Word8, Stream), [(Word8, Stream)])+        doInsert n (s@(n', _) : rest) | n == n' =+            fmap (s:) <$> doInsert (n + 1) rest+        doInsert n streams | n < 63 = lift $ do+            sState <- newTVar StreamOpening+            (sFlowIn, sFlowOut) <- newFlow+            sNextSequence <- newTVar 0+            sWaitingForAck <- newTVar 0+            let info = (n, Stream {..})+            return (info, info : streams)+        doInsert _ _ = throwError "all outbound streams in use" +    runExceptT $ do+        ((streamNumber, stream), outStreams') <- doInsert 1 outStreams+        lift $ writeTVar cOutStreams outStreams'+        return (StreamOpen streamNumber, sFlowIn stream, go cGlobalState streamNumber stream)++  where+    go gs@GlobalState {..} streamNumber stream = do+            (reserved, msg) <- atomically $ do+                readTVar (sState stream) >>= \case+                    StreamRunning -> return ()+                    _             -> retry+                (,) <$> reservePacket conn+                    <*> readFlow (sFlowOut stream)++            (plain, cont, onAck) <- case msg of+                StreamData {..} -> do+                    return (stpData, True, return ())+                StreamClosed {} -> do+                    atomically $ do+                        -- wait for ack on all sent stream data+                        waits <- readTVar (sWaitingForAck stream)+                        when (waits > 0) retry+                    return (BC.empty, False, streamClosed conn streamNumber)++            let secure = True+                plainAckedBy = []+                mbReserved = Just reserved++            mbch <- atomically (readTVar cChannel) >>= return . \case+                ChannelEstablished ch   -> Just ch+                ChannelOurAccept _ _ ch -> Just ch+                _                       -> Nothing++            mbs <- case mbch of+                Just ch -> do+                    runExceptT (channelEncrypt ch $ B.concat+                            [ B.singleton streamNumber+                            , B.singleton (fromIntegral (stpSequence msg) :: Word8)+                            , plain+                            ] ) >>= \case+                        Right (ctext, counter) -> do+                            let isAcked = True+                            return $ Just (0x80 `B.cons` ctext, if isAcked then [ AcknowledgedSingle $ fromIntegral counter ] else [])+                        Left err -> do atomically $ gLog $ "Failed to encrypt data: " ++ err+                                       return Nothing+                Nothing | secure    -> return Nothing+                        | otherwise -> return $ Just (plain, plainAckedBy)++            case mbs of+                Just (bs, ackedBy) -> do+                    atomically $ do+                        modifyTVar' (sWaitingForAck stream) (+ 1)+                    let mbReserved' = (\rs -> rs+                            { rsAckedBy = guard (not $ null ackedBy) >> Just (`elem` ackedBy)+                            , rsOnAck = do+                                rsOnAck rs+                                onAck+                                atomically $ modifyTVar' (sWaitingForAck stream) (subtract 1)+                            }) <$> mbReserved+                    sendBytes conn mbReserved' bs+                Nothing -> return ()++            when cont $ go gs streamNumber stream++connAddReadStream :: Connection addr -> Word8 -> STM RawStreamReader+connAddReadStream Connection {..} streamNumber = do+    inStreams <- readTVar cInStreams+    let doInsert (s@(n, _) : rest)+            | streamNumber <  n = fmap (s:) <$> doInsert rest+            | streamNumber == n = doInsert rest+        doInsert streams = do+            sState <- newTVar StreamRunning+            (sFlowIn, sFlowOut) <- newFlow+            sNextSequence <- newTVar 0+            sWaitingForAck <- newTVar 0+            let stream = Stream {..}+            return (stream, (streamNumber, stream) : streams)+    (stream, inStreams') <- doInsert inStreams+    writeTVar cInStreams inStreams'+    return $ sFlowOut stream+++type RawStreamReader = Flow StreamPacket Void+type RawStreamWriter = Flow Void StreamPacket++data Stream = Stream+    { sState :: TVar StreamState+    , sFlowIn :: Flow Void StreamPacket+    , sFlowOut :: Flow StreamPacket Void+    , sNextSequence :: TVar Word64+    , sWaitingForAck :: TVar Word64+    }++data StreamState = StreamOpening | StreamRunning++data StreamPacket+    = StreamData+        { stpSequence :: Word64+        , stpData :: BC.ByteString+        }+    | StreamClosed+        { stpSequence :: Word64+        }++streamAccepted :: Connection addr -> Word8 -> IO ()+streamAccepted Connection {..} snum = atomically $ do+    (lookup snum <$> readTVar cOutStreams) >>= \case+        Just Stream {..} -> do+            modifyTVar' sState $ \case+                StreamOpening -> StreamRunning+                x             -> x+        Nothing -> return ()++streamClosed :: Connection addr -> Word8 -> IO ()+streamClosed Connection {..} snum = atomically $ do+    modifyTVar' cOutStreams $ filter ((snum /=) . fst)++readStreamToList :: RawStreamReader -> IO (Word64, [(Word64, BC.ByteString)])+readStreamToList stream = readFlowIO stream >>= \case+    StreamData sq bytes -> fmap ((sq, bytes) :) <$> readStreamToList stream+    StreamClosed sqEnd  -> return (sqEnd, [])++readObjectsFromStream :: PartialStorage -> RawStreamReader -> IO (Except String [PartialObject])+readObjectsFromStream st stream = do+    (seqEnd, list) <- readStreamToList stream+    let validate s ((s', bytes) : rest)+            | s == s'   = (bytes : ) <$> validate (s + 1) rest+            | s >  s'   = validate s rest+            | otherwise = throwError "missing object chunk"+        validate s []+            | s == seqEnd = return []+            | otherwise = throwError "content length mismatch"+    return $ do+        content <- BL.fromChunks <$> validate 0 list+        deserializeObjects st content++writeByteStringToStream :: RawStreamWriter -> BL.ByteString -> IO ()+writeByteStringToStream stream = go 0+  where+    go seqNum bstr+        | BL.null bstr = writeFlowIO stream $ StreamClosed seqNum+        | otherwise    = do+            let (cur, rest) = BL.splitAt 500 bstr -- TODO: MTU+            writeFlowIO stream $ StreamData seqNum (BL.toStrict cur)+            go (seqNum + 1) rest++ data WaitingRef = WaitingRef     { wrefStorage :: Storage     , wrefPartial :: PartialRef@@ -195,11 +374,16 @@                   | ChannelOurAccept (Maybe Cookie) (Stored ChannelAccept) Channel                   | ChannelEstablished Channel +data ReservedToSend = ReservedToSend+    { rsAckedBy :: Maybe (TransportHeaderItem -> Bool)+    , rsOnAck :: IO ()+    }  data SentPacket = SentPacket     { spTime :: TimeSpec     , spRetryCount :: Int     , spAckedBy :: Maybe (TransportHeaderItem -> Bool)+    , spOnAck :: IO ()     , spData :: BC.ByteString     } @@ -260,15 +444,19 @@     find ((addr==) . cAddress) <$> readTVar gConnections  newConnection :: GlobalState addr -> addr -> STM (Connection addr)-newConnection GlobalState {..} addr = do+newConnection cGlobalState@GlobalState {..} addr = do     conns <- readTVar gConnections      let cAddress = addr     (cDataUp, cDataInternal) <- newFlow     cChannel <- newTVar ChannelNone     cSecureOutQueue <- newTQueue+    cMaxInFlightPackets <- newTVar 4+    cReservedPackets <- newTVar 0     cSentPackets <- newTVar []     cToAcknowledge <- newTVar []+    cInStreams <- newTVar []+    cOutStreams <- newTVar []     let conn = Connection {..}      writeTVar gConnections (conn : conns)@@ -306,42 +494,73 @@                         case B.uncons dec of                             Just (0x00, content) -> do                                 objs <- deserialize content-                                return (True, objs, Just counter)+                                return $ Left (True, objs, Just counter) +                            Just (snum, dec')+                                | snum < 64+                                , Just (seq8, content) <- B.uncons dec'+                                -> do+                                    return $ Right (snum, seq8, content, counter)+                             Just (_, _) -> do-                                throwError "streams not implemented"+                                throwError "unexpected stream header"                              Nothing -> do                                 throwError "empty decrypted content"                      | b .&. 0xE0 == 0x60 -> do                         objs <- deserialize msg-                        return (False, objs, Nothing)+                        return $ Left (False, objs, Nothing)                      | otherwise -> throwError "invalid packet"                  Nothing -> throwError "empty packet"          runExceptT parse >>= \case-            Right (secure, objs, mbcounter)+            Right (Left (secure, objs, mbcounter))                 | hobj:content <- objs                 , Just header@(TransportHeader items) <- transportFromObject hobj                 -> processPacket gs (maybe (Left addr) Right mbconn) secure (TransportPacket header content) >>= \case-                    Just (conn@Connection {..}, mbup) -> atomically $ do-                        case mbcounter of-                            Just counter | any isHeaderItemAcknowledged items ->-                                modifyTVar' cToAcknowledge (fromIntegral counter :)-                            _ -> return ()-                        processAcknowledgements gs conn items-                        case mbup of-                            Just up -> putTMVar gNextUp (conn, (secure, up))-                            Nothing -> return ()+                    Just (conn@Connection {..}, mbup) -> do+                        ioAfter <- atomically $ do+                            case mbcounter of+                                Just counter | any isHeaderItemAcknowledged items ->+                                    modifyTVar' cToAcknowledge (fromIntegral counter :)+                                _ -> return ()+                            case mbup of+                                Just up -> putTMVar gNextUp (conn, (secure, up))+                                Nothing -> return ()+                            processAcknowledgements gs conn items+                        ioAfter                     Nothing -> return ()                  | otherwise -> atomically $ do                       gLog $ show addr ++ ": invalid objects"                       gLog $ show objs +            Right (Right (snum, seq8, content, counter))+                | Just Connection {..} <- mbconn+                -> atomically $ do+                    (lookup snum <$> readTVar cInStreams) >>= \case+                        Nothing ->+                            gLog $ "unexpected stream number " ++ show snum++                        Just Stream {..} -> do+                            expectedSequence <- readTVar sNextSequence+                            let seqFull = expectedSequence - 0x80 + fromIntegral (seq8 - fromIntegral expectedSequence + 0x80 :: Word8)+                            sdata <- if+                                | B.null content -> do+                                    modifyTVar' cInStreams $ filter ((/=snum) . fst)+                                    return $ StreamClosed seqFull+                                | otherwise -> do+                                    writeTVar sNextSequence $ max expectedSequence (seqFull + 1)+                                    return $ StreamData seqFull content+                            writeFlow sFlowIn sdata+                            modifyTVar' cToAcknowledge (fromIntegral counter :)++                | otherwise -> do+                    atomically $ gLog $ show addr <> ": stream packet without connection"+             Left err -> do                 atomically $ gLog $ show addr <> ": failed to parse packet: " <> err @@ -455,10 +674,27 @@ verifyCookie :: GlobalState addr -> addr -> Cookie -> IO Bool verifyCookie GlobalState {} addr (Cookie cookie) = return $ show addr == BC.unpack cookie -resendBytes :: GlobalState addr -> Connection addr -> SentPacket -> IO ()-resendBytes GlobalState {..} Connection {..} sp = do++reservePacket :: Connection addr -> STM ReservedToSend+reservePacket Connection {..} = do+    maxPackets <- readTVar cMaxInFlightPackets+    reserved <- readTVar cReservedPackets+    sent <- length <$> readTVar cSentPackets++    when (sent + reserved >= maxPackets) $ do+        retry++    writeTVar cReservedPackets $ reserved + 1+    return $ ReservedToSend Nothing (return ())++resendBytes :: Connection addr -> Maybe ReservedToSend -> SentPacket -> IO ()+resendBytes Connection {..} reserved sp = do+    let GlobalState {..} = cGlobalState     now <- getTime MonotonicRaw     atomically $ do+        when (isJust reserved) $ do+            modifyTVar' cReservedPackets (subtract 1)+         when (isJust $ spAckedBy sp) $ do             modifyTVar' cSentPackets $ (:) sp                 { spTime = now@@ -466,12 +702,13 @@                 }         writeFlow gDataFlow (cAddress, spData sp) -sendBytes :: GlobalState addr -> Connection addr -> ByteString -> Maybe (TransportHeaderItem -> Bool) -> IO ()-sendBytes gs conn bs ackedBy = resendBytes gs conn+sendBytes :: Connection addr -> Maybe ReservedToSend -> ByteString -> IO ()+sendBytes conn reserved bs = resendBytes conn reserved     SentPacket         { spTime = undefined         , spRetryCount = -1-        , spAckedBy = ackedBy+        , spAckedBy = rsAckedBy =<< reserved+        , spOnAck = maybe (return ()) rsOnAck reserved         , spData = bs         } @@ -486,18 +723,22 @@                     _                     -> Nothing              let checkOutstanding-                    | isJust mbch = readTQueue cSecureOutQueue+                    | isJust mbch = do+                        (,) <$> readTQueue cSecureOutQueue <*> (Just <$> reservePacket conn)                     | otherwise = retry +                checkDataInternal = do+                    (,) <$> readFlow cDataInternal <*> (Just <$> reservePacket conn)+                 checkAcknowledgements                     | isJust mbch = do                         acks <- readTVar cToAcknowledge                         if null acks then retry-                                     else return (True, TransportPacket (TransportHeader []) [], [])+                                     else return ((True, TransportPacket (TransportHeader []) [], []), Nothing)                     | otherwise = retry -            (secure, packet@(TransportPacket (TransportHeader hitems) content), plainAckedBy) <--                checkOutstanding <|> readFlow cDataInternal <|> checkAcknowledgements+            ((secure, packet@(TransportPacket (TransportHeader hitems) content), plainAckedBy), mbReserved) <-+                checkOutstanding <|> checkDataInternal <|> checkAcknowledgements              when (isNothing mbch && secure) $ do                 writeTQueue cSecureOutQueue (secure, packet, plainAckedBy)@@ -514,6 +755,10 @@                         (serializeObject $ transportToObject gStorage header)                         : map lazyLoadBytes content +                let onAck = case catMaybes (map (\case StreamOpen n -> Just n; _ -> Nothing) hitems) of+                                 [] -> return ()+                                 xs -> sequence_ $ map (streamAccepted conn) xs+                 mbs <- case mbch of                     Just ch -> do                         runExceptT (channelEncrypt ch $ BL.toStrict $ 0x00 `BL.cons` plain) >>= \case@@ -526,7 +771,12 @@                             | otherwise -> return $ Just (BL.toStrict plain, plainAckedBy)                  case mbs of-                    Just (bs, ackedBy) -> sendBytes gs conn bs $ guard (not $ null ackedBy) >> Just (`elem` ackedBy)+                    Just (bs, ackedBy) -> do+                        let mbReserved' = (\rs -> rs+                                { rsAckedBy = guard (not $ null ackedBy) >> Just (`elem` ackedBy)+                                , rsOnAck = rsOnAck rs >> onAck+                                }) <$> mbReserved+                        sendBytes conn mbReserved' bs                     Nothing -> return ()      let retransmitPacket :: Connection addr -> STM (IO ())@@ -544,8 +794,9 @@                            return $ return ()                    else retry               else do+                reserved <- reservePacket conn                 writeTVar cSentPackets rest-                return $ resendBytes gs conn sp+                return $ resendBytes conn (Just reserved) sp      let handleControlRequests = readFlow gControlFlow >>= \case             RequestConnection addr -> do@@ -553,6 +804,7 @@                 identity <- fst <$> readTVar gIdentity                 readTVar cChannel >>= \case                     ChannelNone -> do+                        reserved <- reservePacket conn                         let packet = BL.toStrict $ BL.concat                                 [ serializeObject $ transportToObject gStorage $ TransportHeader $                                     [ Initiation $ refDigest gInitConfig@@ -561,7 +813,8 @@                                 , lazyLoadBytes gInitConfig                                 ]                         writeTVar cChannel ChannelCookieWait-                        return $ sendBytes gs conn packet $ Just $ \case CookieSet {} -> True; _ -> False+                        let reserved' = reserved { rsAckedBy = Just $ \case CookieSet {} -> True; _ -> False }+                        return $ sendBytes conn (Just reserved') packet                     _ -> return $ return ()              SendAnnounce addr -> do@@ -584,6 +837,8 @@         , [ handleControlRequests ]         ] -processAcknowledgements :: GlobalState addr -> Connection addr -> [TransportHeaderItem] -> STM ()-processAcknowledgements GlobalState {} Connection {..} = mapM_ $ \hitem -> do-    modifyTVar' cSentPackets $ filter $ \sp -> not (fromJust (spAckedBy sp) hitem)+processAcknowledgements :: GlobalState addr -> Connection addr -> [TransportHeaderItem] -> STM (IO ())+processAcknowledgements GlobalState {} Connection {..} header = do+    (acked, notAcked) <- partition (\sp -> any (fromJust (spAckedBy sp)) header) <$> readTVar cSentPackets+    writeTVar cSentPackets notAcked+    return $ sequence_ $ map spOnAck acked

@@ -4,6 +4,7 @@     emptySet,     loadSet,     storeSetAdd,+    storeSetAddComponent,      fromSetBy, ) where@@ -23,6 +24,7 @@ import Erebos.Util  data Set a = Set [Stored (SetItem (Component a))]+    deriving (Eq)  data SetItem a = SetItem     { siPrev :: [Stored (SetItem a)]@@ -54,6 +56,12 @@ storeSetAdd st x (Set prev) = Set . (:[]) <$> wrappedStore st SetItem     { siPrev = prev     , siItem = toComponents x+    }++storeSetAddComponent :: (Mergeable a, MonadStorage m, MonadIO m) => Stored (Component a) -> Set a -> m (Set a)+storeSetAddComponent component (Set prev) = Set . (:[]) <$> mstore SetItem+    { siPrev = prev+    , siItem = [ component ]     }  

@@ -86,6 +86,8 @@  class (MonadIO m, MonadStorage m) => MonadHead a m where     updateLocalHead :: (Stored a -> m (Stored a, b)) -> m b+    getLocalHead :: m (Stored a)+    getLocalHead = updateLocalHead $ \x -> return (x, x)  updateLocalHead_ :: MonadHead a m => (Stored a -> m (Stored a)) -> m () updateLocalHead_ f = updateLocalHead (fmap (,()) . f)

@@ -10,6 +10,7 @@     generations,     ancestors,     precedes,+    precedesOrEquals,     filterAncestors,     storedRoots,     walkAncestors,@@ -108,6 +109,9 @@  precedes :: Storable a => Stored a -> Stored a -> Bool precedes x y = not $ x `elem` filterAncestors [x, y]++precedesOrEquals :: Storable a => Stored a -> Stored a -> Bool+precedesOrEquals x y = filterAncestors [ x, y ] == [ y ]  filterAncestors :: Storable a => [Stored a] -> [Stored a] filterAncestors [x] = [x]