Hash2Pub.cabal

@@ -46,8 +46,8 @@ category:            Network
 extra-source-files:  CHANGELOG.md
 
 common deps
-  build-depends:       base >=4, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=3.1, time, cmdargs ^>= 0.10, cryptonite, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist, formatting
-  ghc-options:         -Wall -Wpartial-fields -O2
+  build-depends:       base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types
+  ghc-options:         -Wall
 
 
 
@@ -58,7 +58,7 @@ library
   exposed-modules:     Hash2Pub.FediChord, Hash2Pub.FediChordTypes, Hash2Pub.DHTProtocol, Hash2Pub.ASN1Coding, Hash2Pub.ProtocolTypes, Hash2Pub.PostService, Hash2Pub.RingMap
 
   -- Modules included in this library but not exported.
-  other-modules: Hash2Pub.Utils, Hash2Pub.PostService.API
+  other-modules: Hash2Pub.Utils
 
   -- LANGUAGE extensions used by modules in this package.
   other-extensions:    GeneralizedNewtypeDeriving, DataKinds, OverloadedStrings
@@ -91,21 +91,7 @@ executable Hash2Pub
   -- Base language which the package is written in.
   default-language:    Haskell2010
 
-  ghc-options:         -threaded -rtsopts -with-rtsopts=-N
-
-executable Experiment
-  -- experiment runner
-  import: deps
-
-  build-depends: Hash2Pub
-
-  main-is: Experiment.hs
-
-  hs-source-dirs: app
-
-  default-language: Haskell2010
-
-  ghc-options: -threaded
+  ghc-options:         -threaded
 
 
 test-suite Hash2Pub-test

Readme.md

@@ -1,7 +1,7 @@
 # Hash2Pub
 
 ***This is heavily WIP and does not provide any useful functionality yet***.  
-I aim for always having the `mainline` branch in a state where it builds and tests pass.
+I aim for always having the master branch at a state where it builds and tests pass.
 
 A fully-decentralised relay for global hashtag federation in [ActivityPub](https://activitypub.rocks) based on a distributed hash table.
 It allows querying and subscribing to all posts of a certain hashtag and is implemented in Haskell.
@@ -10,13 +10,8 @@ This is the practical implementation of the concept presented in the paper [Dece
 
 The ASN.1 module schema used for DHT messages can be found in `FediChord.asn1`.
 
-For further questions and discussins, please refer to the **Hash2Pub topic in [SocialHub](https://socialhub.activitypub.rocks/c/software/hash2pub/48)**.
-
 ## Building
 
 The project and its developent environment are built with [Nix](https://nixos.org/nix/).
 
-The development environment can be entered with `nix-shell shell-minimal.nix`. Then the project can be built with `cabal build` from within the environment, or using `nix-shell --command "cabal build" shell-minimal.nix` to do both steps at once.
-
-While the `shell-minimal.nix` environment contains everything necessary for building and testing this project, the `shell.nix` additionally contains the Haskell IDE engine *hie* and the documentation for all used Haskell packages for more convenient development.  
-Be aware that these need to be build from source and can take a very long time to build.
+The development environment can be entered with `nix-shell`. Then the project can be built with `cabal build` from within the environment, or using `nix-shell --command "cabal build"` to do both steps at once.

app/Experiment.hs

@@ -1,51 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-
-module Main where
-
-import           Control.Concurrent
-import           Control.Monad        (forM_)
-import qualified Data.Text.Lazy       as Txt
-import qualified Data.Text.Lazy.IO    as TxtI
-import qualified Network.HTTP.Client  as HTTP
-import           System.Environment   (getArgs)
-
-import           Hash2Pub.PostService (Hashtag, clientPublishPost)
-
--- configuration constants
-timelineFile = "../simulationData/inputs/generated/timeline_sample.csv"
-
-main :: IO ()
-main = do
-    -- read CLI parameters
-    speedupStr : _ <- getArgs
-    -- read and parse timeline schedule
-    -- relying on lazyness of HaskellIO, hoping it does not introduce too strong delays
-    postEvents <- parseSchedule <$> TxtI.readFile timelineFile
-    -- actually schedule and send the post events
-    executeSchedule (read speedupStr) postEvents
-    pure ()
-
-
-
-parseSchedule :: Txt.Text
-              -> [(Int, Hashtag, (String, Int))] -- ^ [(delay in microseconds, hashtag, (hostname, port))]
-parseSchedule = fmap (parseEntry . Txt.split (== ';')) . Txt.lines
-  where
-      parseEntry [delayT, contactT, tag] =
-          (read $ Txt.unpack delayT, tag, read $ Txt.unpack contactT)
-      parseEntry entry = error $ "invalid schedule input format: " <> show entry
-
-executeSchedule :: Int  -- ^ speedup factor
-                -> [(Int, Hashtag, (String, Int))] -- ^ [(delay in microseconds, hashtag, (hostname, port))]
-                -> IO ()
-executeSchedule speedup events = do
-    -- initialise HTTP manager
-    httpMan <- HTTP.newManager $ HTTP.defaultManagerSettings { HTTP.managerResponseTimeout = HTTP.responseTimeoutMicro 60000000 }
-    forM_ events $ \(delay, tag, (pubHost, pubPort)) -> do
-        _ <- forkIO $
-            clientPublishPost httpMan pubHost pubPort ("foobar #" <> tag)
-            >>= either putStrLn (const $ pure ())
-        -- while threadDelay gives only minimum delay guarantees, let's hope the
-        -- additional delays are negligible
-        -- otherwise: evaluate usage of https://hackage.haskell.org/package/schedule-0.3.0.0/docs/Data-Schedule.html
-        threadDelay $ delay `div` speedup

app/Main.hs

@@ -45,36 +45,21 @@ main = do
 
 readConfig :: IO (FediChordConf, ServiceConf)
 readConfig = do
-    confDomainString : ipString : portString : servicePortString : speedupString : remainingArgs <- getArgs
-    -- allow starting the initial node without bootstrapping info to avoid
-    -- waiting for timeout
+    confDomainString : ipString : portString : bootstrapHost : bootstrapPortString : servicePortString : speedup : _ <- getArgs
     let
-        speedup = read speedupString
-        confBootstrapNodes' = case remainingArgs of
-            bootstrapHost : bootstrapPortString : _ ->
-                [(bootstrapHost, read bootstrapPortString)]
-            _ -> []
-        fConf = FediChordConf
-          { confDomain = confDomainString
-          , confIP = toHostAddress6 . read $ ipString
-          , confDhtPort = read portString
-          , confBootstrapNodes = confBootstrapNodes'
-          , confStabiliseInterval = 80 * 10^6
-          , confBootstrapSamplingInterval = 180 * 10^6 `div` speedup
-          , confMaxLookupCacheAge = 300 / fromIntegral speedup
-          , confJoinAttemptsInterval = 60 * 10^6 `div` speedup
-          , confMaxNodeCacheAge = 600 / fromIntegral speedup
-          , confResponsePurgeAge = 60 / fromIntegral speedup
-          , confRequestTimeout = 5 * 10^6 `div` speedup
-          , confRequestRetries = 3
-          }
-        sConf = ServiceConf
-          { confSubscriptionExpiryTime = 24*3600 / fromIntegral speedup
-          , confServicePort = read servicePortString
-          , confServiceHost = confDomainString
-          , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log"
-          , confSpeedupFactor = speedup
-          , confStatsEvalDelay = 120 * 10^6 `div` speedup
-          }
+        fConf = FediChordConf {
+        confDomain = confDomainString
+      , confIP = toHostAddress6 . read $ ipString
+      , confDhtPort = read portString
+      , confBootstrapNodes = [(bootstrapHost, read bootstrapPortString)]
+      --, confStabiliseInterval = 60
+      , confBootstrapSamplingInterval = 180
+      , confMaxLookupCacheAge = 300
+                           }
+        sConf = ServiceConf {
+        confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` (read speedup :: Integer)
+                            , confServicePort = read servicePortString
+                            , confServiceHost = confDomainString
+                            }
     pure (fConf, sConf)
 

default.nix

@@ -1,18 +1,26 @@
 {
-  compiler ? "ghc884"
+  compiler ? "ghc865",
+  withHIE ? false
 }:
 
 let
+  # pin all-hies for getting the language server
+  all-hies = fetchTarball {
+    url = "https://github.com/infinisil/all-hies/tarball/b8fb659620b99b4a393922abaa03a1695e2ca64d";
+    sha256 = "sha256:0br6wsqpfk1lzz90f7zw439w1ir2p54268qilw9l2pk6yz7ganfx";
+  };
   pkgs = import (
     builtins.fetchGit {
       name = "nixpkgs-pinned";
       url = https://github.com/NixOS/nixpkgs/;
-      ref = "refs/heads/release-20.09";
-      rev = "e065200fc90175a8f6e50e76ef10a48786126e1c";
+      ref = "refs/heads/release-20.03";
+      rev = "076c67fdea6d0529a568c7d0e0a72e6bc161ecf5";
     }) {
       # Pass no config for purity
           config = {};
-              overlays = [];
+              overlays = [
+                      (import all-hies {}).overlay
+                          ];
     };
   hp = pkgs.haskell.packages."${compiler}";
   src = pkgs.nix-gitignore.gitignoreSource [] ./.;
@@ -30,6 +38,7 @@ in
         hlint
         stylish-haskell
         pkgs.python3Packages.asn1ate
-      ];
+      ]
+      ++ (if withHIE then [ hie ] else []);
   };
 }

shell.nix

@@ -1 +1 @@
-(import ./default.nix {}).shell
+(import ./default.nix {withHIE = true;}).shell

src/Hash2Pub/DHTProtocol.hs

@@ -49,11 +49,9 @@ import           Control.Concurrent.STM.TQueue
 import           Control.Concurrent.STM.TVar
 import           Control.Exception
 import           Control.Monad                  (foldM, forM, forM_, void, when)
-import           Control.Monad.Except           (MonadError (..), runExceptT)
-import           Control.Monad.IO.Class         (MonadIO (..))
 import qualified Data.ByteString                as BS
 import           Data.Either                    (rights)
-import           Data.Foldable                  (foldl', foldr', foldrM)
+import           Data.Foldable                  (foldl', foldr')
 import           Data.Functor.Identity
 import           Data.IP                        (IPv6, fromHostAddress6,
                                                  toHostAddress6)
@@ -108,6 +106,9 @@ queryLocalCache ownState nCache lBestNodes targetID
     -- the closest succeeding node (like with the p initiated parallel queries
       | otherwise = FORWARD $ closestSuccessor `Set.union` closestCachePredecessors (lBestNodes-1) targetID nCache
   where
+    ownID = getNid ownState
+    preds = predecessors ownState
+
     closestSuccessor :: Set.Set RemoteCacheEntry
     closestSuccessor = maybe Set.empty (Set.singleton . toRemoteCacheEntry) $ cacheLookupSucc targetID nCache
 
@@ -258,6 +259,7 @@ handleIncomingRequest :: Service s (RealNodeSTM s)
                       -> SockAddr                           -- ^ source address of the request
                       -> IO ()
 handleIncomingRequest nsSTM sendQ msgSet sourceAddr = do
+    putStrLn $ "handling incoming request: " <> show msgSet
     ns <- readTVarIO nsSTM
     -- add nodestate to cache
     now <- getPOSIXTime
@@ -312,6 +314,7 @@ handleIncomingRequest nsSTM sendQ msgSet sourceAddr = do
 -- | execute a key ID lookup on local cache and respond with the result
 respondQueryID :: LocalNodeStateSTM s -> Set.Set FediChordMessage -> IO (Map.Map Integer BS.ByteString)
 respondQueryID nsSTM msgSet = do
+    putStrLn "responding to a QueryID request"
     -- this message cannot be split reasonably, so just
     -- consider the first payload
     let
@@ -432,9 +435,7 @@ respondJoin nsSTM msgSet = do
         let
             aRequestPart = Set.elemAt 0 msgSet
             senderNS = sender aRequestPart
-            -- if not joined yet, attract responsibility for
-            -- all keys to make bootstrapping possible
-            responsibilityLookup = if isJoined nsSnap then queryLocalCache nsSnap cache 1 (getNid senderNS) else FOUND (toRemoteNodeState nsSnap)
+            responsibilityLookup = queryLocalCache nsSnap cache 1 (getNid senderNS)
             thisNodeResponsible (FOUND _)   = True
             thisNodeResponsible (FORWARD _) = False
         -- check whether the joining node falls into our responsibility
@@ -488,11 +489,10 @@ requestJoin :: (NodeState a, Service s (RealNodeSTM s)) => a             -- ^ cu
 requestJoin toJoinOn ownStateSTM = do
     ownState <- readTVarIO ownStateSTM
     prn <- readTVarIO $ parentRealNode ownState
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ownState)
-    let srcAddr = confIP nodeConf
+    srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ownState)
     bracket (mkSendSocket srcAddr (getDomain toJoinOn) (getDhtPort toJoinOn)) close (\sock -> do
         -- extract own state for getting request information
-        responses <- sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid -> Request rid (toRemoteNodeState ownState) 1 True Join (Just JoinRequestPayload)) sock
+        responses <- sendRequestTo 5000 3 (\rid -> Request rid (toRemoteNodeState ownState) 1 True Join (Just JoinRequestPayload)) sock
         (cacheInsertQ, joinedState) <- atomically $ do
             stateSnap <- readTVar ownStateSTM
             let
@@ -517,28 +517,28 @@ requestJoin toJoinOn ownStateSTM = do
                     ([], Set.empty, Set.empty)
                     responses
                 -- sort, slice and set the accumulated successors and predecessors
-                -- the contacted node itself is a successor as well and, with few
-                -- nodes, can be a predecessor as well
-                newState = setSuccessors (toRemoteNodeState toJoinOn:Set.elems succAccSet) . setPredecessors (toRemoteNodeState toJoinOn:Set.elems predAccSet) $ stateSnap
+                newState = setSuccessors (Set.elems succAccSet) . setPredecessors (Set.elems predAccSet) $ stateSnap
             writeTVar ownStateSTM newState
             pure (cacheInsertQ, newState)
         -- execute the cache insertions
         mapM_ (\f -> f joinedState) cacheInsertQ
         if responses == Set.empty
                   then pure . Left $ "join error: got no response from " <> show (getNid toJoinOn)
-           else do
-                -- wait for migration data to be completely received
-                waitForMigrationFrom (nodeService prn) (getNid toJoinOn)
-                pure $ Right ownStateSTM
+           else if null (predecessors joinedState) && null (successors joinedState)
+                then pure $ Left "join error: no predecessors or successors"
+                -- successful join
+                else do
+                    -- wait for migration data to be completely received
+                    waitForMigrationFrom (nodeService prn) (getNid ownState)
+                    pure $ Right ownStateSTM
                                    )
         `catch` (\e -> pure . Left $ displayException (e :: IOException))
 
 
 -- | Send a 'QueryID' 'Request' for getting the node that handles a certain key ID.
-requestQueryID :: (MonadIO m, MonadError String m)
-               => LocalNodeState s        -- ^ NodeState of the querying node
+requestQueryID :: LocalNodeState s        -- ^ NodeState of the querying node
                -> NodeID                -- ^ target key ID to look up
-               -> m RemoteNodeState    -- ^ the node responsible for handling that key
+               -> IO RemoteNodeState    -- ^ the node responsible for handling that key
 -- 1. do a local lookup for the l closest nodes
 -- 2. create l sockets
 -- 3. send a message async concurrently to all l nodes
@@ -546,23 +546,23 @@ requestQueryID :: (MonadIO m, MonadError String m)
 -- 5. repeat until FOUND (problem: new entries not necessarily already in cache, explicitly compare with closer results)
 -- TODO: deal with lookup failures
 requestQueryID ns targetID = do
-    firstCacheSnapshot <- liftIO . readTVarIO . nodeCacheSTM $ ns
+    firstCacheSnapshot <- readTVarIO . nodeCacheSTM $ ns
     -- TODO: make maxAttempts configurable
     queryIdLookupLoop firstCacheSnapshot ns 50 targetID
 
 -- | like 'requestQueryID, but allows passing of a custom cache, e.g. for joining
-queryIdLookupLoop :: (MonadIO m, MonadError String m) => NodeCache -> LocalNodeState s -> Int -> NodeID -> m RemoteNodeState
+queryIdLookupLoop :: NodeCache -> LocalNodeState s -> Int -> NodeID -> IO RemoteNodeState
 -- return node itself as default fallback value against infinite recursion.
 -- TODO: consider using an Either instead of a default value
-queryIdLookupLoop _ ns 0 _ = throwError "exhausted maximum lookup attempts"
+queryIdLookupLoop _ ns 0 _ = pure $ toRemoteNodeState ns
 queryIdLookupLoop cacheSnapshot ns maxAttempts targetID = do
     let localResult = queryLocalCache ns cacheSnapshot (lNumBestNodes ns) targetID
     -- FOUND can only be returned if targetID is owned by local node
     case localResult of
       FOUND thisNode -> pure thisNode
       FORWARD nodeSet -> do
-          responseEntries <- liftIO $ sendQueryIdMessages targetID ns Nothing (remoteNode <$> Set.elems nodeSet)
-          now <- liftIO getPOSIXTime
+          responseEntries <- sendQueryIdMessages targetID ns Nothing (remoteNode <$> Set.elems nodeSet)
+          now <- getPOSIXTime
           -- check for a FOUND and return it
           case responseEntries of
             FOUND foundNode -> pure foundNode
@@ -585,10 +585,10 @@ sendQueryIdMessages targetID ns lParam targets = do
 
           -- create connected sockets to all query targets and use them for request handling
 
-          nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns)
-          let srcAddr = confIP nodeConf
+          srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns)
+          -- ToDo: make attempts and timeout configurable
           queryThreads <- mapM (\resultNode -> async $ bracket (mkSendSocket srcAddr (getDomain resultNode) (getDhtPort resultNode)) close (
-              sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (lookupMessage targetID ns Nothing)
+              sendRequestTo 5000 3 (lookupMessage targetID ns Nothing)
                                                                                                                                    )) targets
           -- ToDo: process results immediately instead of waiting for the last one to finish, see https://stackoverflow.com/a/38815224/9198613
           -- ToDo: exception handling, maybe log them
@@ -596,10 +596,8 @@ sendQueryIdMessages targetID ns lParam targets = do
           -- insert new cache entries both into global cache as well as return accumulated QueryResponses for further processing
           now <- getPOSIXTime
           -- collect cache entries from all responses
-          foldrM (\resp acc -> do
-            let
-                responseResult = queryResult <$> payload resp
-                entrySet = case responseResult of
+          foldM (\acc resp -> do
+            let entrySet = case queryResult <$> payload resp of
                              Just (FOUND result1) -> Set.singleton (RemoteCacheEntry result1 now)
                              Just (FORWARD resultset) -> resultset
                              _ -> Set.empty
@@ -609,15 +607,10 @@ sendQueryIdMessages targetID ns lParam targets = do
             -- return accumulated QueryResult
             pure $ case acc of
               -- once a FOUND as been encountered, return this as a result
-              FOUND{} -> acc
-              FORWARD accSet
-                | maybe False isFound responseResult -> fromJust responseResult
-                | otherwise -> FORWARD $ entrySet `Set.union` accSet
+              isFound@FOUND{} -> isFound
+              FORWARD accSet  -> FORWARD $ entrySet `Set.union` accSet
 
                             ) (FORWARD Set.empty) responses
-  where
-      isFound FOUND{} = True
-      isFound _       = False
 
 -- | Create a QueryID message to be supplied to 'sendRequestTo'
 lookupMessage :: Integral i
@@ -636,9 +629,8 @@ requestStabilise :: LocalNodeState s      -- ^ sending node
                  -> RemoteNodeState     -- ^ neighbour node to send to
                  -> IO (Either String ([RemoteNodeState], [RemoteNodeState]))   -- ^ (predecessors, successors) of responding node
 requestStabilise ns neighbour = do
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns)
-    let srcAddr = confIP nodeConf
-    responses <- bracket (mkSendSocket srcAddr (getDomain neighbour) (getDhtPort neighbour)) close (fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid ->
+    srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns)
+    responses <- bracket (mkSendSocket srcAddr (getDomain neighbour) (getDhtPort neighbour)) close (fmap Right . sendRequestTo 5000 3 (\rid ->
         Request {
             requestID = rid
           , sender = toRemoteNodeState ns
@@ -660,7 +652,8 @@ requestStabilise ns neighbour = do
                                                       )
                                                       ([],[]) respSet
             -- update successfully responded neighbour in cache
-            maybe (pure ()) (\p -> queueUpdateVerifieds (Identity $ senderID p) ns) $ headMay (Set.elems respSet)
+            now <- getPOSIXTime
+            maybe (pure ()) (\p -> queueAddEntries (Identity $ RemoteCacheEntry (sender p) now) ns) $ headMay (Set.elems respSet)
             pure $ if null responsePreds && null responseSuccs
                       then Left "no neighbours returned"
                       else Right (responsePreds, responseSuccs)
@@ -675,14 +668,13 @@ requestLeave :: LocalNodeState s
              -> RemoteNodeState     -- target node
              -> IO (Either String ())   -- error or success
 requestLeave ns doMigration target = do
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns)
-    let srcAddr = confIP nodeConf
-        leavePayload = LeaveRequestPayload {
+    srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns)
+    let leavePayload = LeaveRequestPayload {
         leaveSuccessors = successors ns
       , leavePredecessors = predecessors ns
       , leaveDoMigration = doMigration
                                            }
-    responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close (fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid ->
+    responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close (fmap Right . sendRequestTo 5000 3 (\rid ->
         Request {
             requestID = rid
           , sender = toRemoteNodeState ns
@@ -704,11 +696,10 @@ requestPing :: LocalNodeState s      -- ^ sending node
                  -> RemoteNodeState     -- ^ node to be PINGed
                  -> IO (Either String [RemoteNodeState])   -- ^ all active vServers of the pinged node
 requestPing ns target = do
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns)
-    let srcAddr = confIP nodeConf
+    srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns)
     responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close
         (\sock -> do
-            resp <- sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid ->
+            resp <- sendRequestTo 5000 3 (\rid ->
                 Request {
                     requestID = rid
                   , sender = toRemoteNodeState ns
@@ -744,9 +735,10 @@ requestPing ns target = do
         ) responses
 
 
+
 -- | Generic function for sending a request over a connected socket and collecting the response.
 -- Serialises the message and tries to deliver its parts for a number of attempts within a specified timeout.
-sendRequestTo :: Int                    -- ^ timeout in milliseconds
+sendRequestTo :: Int                    -- ^ timeout in seconds
               -> Int                    -- ^ number of retries
               -> (Integer -> FediChordMessage)       -- ^ the message to be sent, still needing a requestID
               -> Socket                 -- ^ connected socket to use for sending
@@ -757,10 +749,11 @@ sendRequestTo timeoutMillis numAttempts msgIncomplete sock = do
     let
         msgComplete = msgIncomplete randomID
         requests = serialiseMessage sendMessageSize msgComplete
+    putStrLn $ "sending request message " <> show msgComplete
     -- create a queue for passing received response messages back, even after a timeout
     responseQ <- newTBQueueIO $ 2*maximumParts  -- keep room for duplicate packets
     -- start sendAndAck with timeout
-    _ <- attempts numAttempts . timeout (timeoutMillis*1000) $ sendAndAck responseQ sock requests
+    attempts numAttempts . timeout timeoutMillis $ sendAndAck responseQ sock requests
     -- after timeout, check received responses, delete them from unacked message set/ map and rerun senAndAck with that if necessary.
     recvdParts <- atomically $ flushTBQueue responseQ
     pure $ Set.fromList recvdParts
@@ -769,20 +762,19 @@ sendRequestTo timeoutMillis numAttempts msgIncomplete sock = do
                -> Socket                            -- ^ the socket used for sending and receiving for this particular remote node
                -> Map.Map Integer BS.ByteString     -- ^ the remaining unacked request parts
                -> IO ()
-    sendAndAck responseQueue sock' remainingSends = do
-        sendMany sock' $ Map.elems remainingSends
+    sendAndAck responseQueue sock remainingSends = do
+        sendMany sock $ Map.elems remainingSends
         -- if all requests have been acked/ responded to, return prematurely
-        recvLoop sock' responseQueue remainingSends Set.empty Nothing
-    recvLoop :: Socket
-             -> TBQueue FediChordMessage          -- ^ the queue for putting in the received responses
+        recvLoop responseQueue remainingSends Set.empty Nothing
+    recvLoop :: TBQueue FediChordMessage          -- ^ the queue for putting in the received responses
              -> Map.Map Integer BS.ByteString       -- ^ the remaining unacked request parts
              -> Set.Set Integer                     -- ^ already received response part numbers
-             -> Maybe Integer                     -- ^ total number of response parts if already known
+               -> Maybe Integer                     -- ^ total number of response parts if already known
              -> IO ()
-    recvLoop sock' responseQueue remainingSends' receivedPartNums totalParts = do
+    recvLoop responseQueue remainingSends' receivedPartNums totalParts = do
         -- 65535 is maximum length of UDP packets, as long as
         -- no IPv6 jumbograms are used
-        response <- deserialiseMessage <$> recv sock' 65535
+        response <- deserialiseMessage <$> recv sock 65535
         case response of
           Right msg@Response{} -> do
               atomically $ writeTBQueue responseQueue msg
@@ -790,12 +782,11 @@ sendRequestTo timeoutMillis numAttempts msgIncomplete sock = do
                 newTotalParts = if isFinalPart msg then Just (part msg) else totalParts
                 newRemaining = Map.delete (part msg) remainingSends'
                 newReceivedParts = Set.insert (part msg) receivedPartNums
-              if  Map.null newRemaining && maybe False (\p -> Set.size newReceivedParts == fromIntegral p) newTotalParts
+              if Map.null newRemaining && maybe False (\p -> Set.size receivedPartNums == fromIntegral p) newTotalParts
                  then pure ()
-                 else recvLoop sock' responseQueue newRemaining newReceivedParts newTotalParts
+                 else recvLoop responseQueue newRemaining receivedPartNums newTotalParts
           -- drop errors and invalid messages
-          Right Request{} -> pure ()    -- expecting a response, not a request
-          Left _ -> recvLoop sock' responseQueue remainingSends' receivedPartNums totalParts
+          Left _ -> recvLoop responseQueue remainingSends' receivedPartNums totalParts
 
 
 -- | enqueue a list of RemoteCacheEntries to be added to the global NodeCache
@@ -821,18 +812,6 @@ queueDeleteEntry :: NodeID
                  -> IO ()
 queueDeleteEntry toDelete = queueDeleteEntries $ Identity toDelete
 
-
--- | enqueue the timestamp update and verification marking of an entry in the
--- global 'NodeCache'.
-queueUpdateVerifieds :: Foldable c
-                     => c NodeID
-                     -> LocalNodeState s
-                     -> IO ()
-queueUpdateVerifieds nIds ns = do
-    now <- getPOSIXTime
-    forM_ nIds $ \nid' -> atomically $ writeTQueue (cacheWriteQueue ns) $
-        markCacheEntryAsVerified (Just now) nid'
-
 -- | retry an IO action at most *i* times until it delivers a result
 attempts :: Int             -- ^ number of retries *i*
          -> IO (Maybe a)    -- ^ action to retry
@@ -865,7 +844,7 @@ mkServerSocket ip port = do
     sockAddr <- addrAddress <$> resolve (Just $ show . fromHostAddress6 $ ip) (Just port)
     sock <- socket AF_INET6 Datagram defaultProtocol
     setSocketOption sock IPv6Only 1
-    bind sock sockAddr `catch` (\e -> putStrLn $ "Caught exception while bind " <> show sock <> " " <> show sockAddr <> ": " <> show (e :: SomeException))
+    bind sock sockAddr
     pure sock
 
 -- | create a UDP datagram socket, connected to a destination.
@@ -881,6 +860,6 @@ mkSendSocket srcIp dest destPort = do
     setSocketOption sendSock IPv6Only 1
     -- bind to the configured local IP to make sure that outgoing packets are sent from
     -- this source address
-    bind sendSock srcAddr `catch` (\e -> putStrLn $ "Caught exception while mkSendSocket bind " <> show sendSock <> " " <> show srcAddr <> ": " <> show (e :: SomeException))
+    bind sendSock srcAddr
     connect sendSock destAddr
     pure sendSock

src/Hash2Pub/FediChord.hs

@@ -166,7 +166,6 @@ fediChordBootstrapJoin nsSTM bootstrapNode = do
         currentlyResponsible <- liftEither lookupResp
         liftIO . putStrLn $ "Trying to join on " <> show (getNid currentlyResponsible)
         -- 2. then send a join to the currently responsible node
-        liftIO $ putStrLn "send a bootstrap Join"
         joinResult <- liftIO $ requestJoin currentlyResponsible nsSTM
         liftEither joinResult
 
@@ -199,7 +198,7 @@ convergenceSampleThread nsSTM = forever $ do
     -- unjoined node: try joining through all bootstrapping nodes
     else tryBootstrapJoining nsSTM >> pure ()
     let delaySecs = confBootstrapSamplingInterval . nodeConfig $ parentNode
-    threadDelay delaySecs
+    threadDelay $ delaySecs * 10^6
 
 
 -- | Try joining the DHT through any of the bootstrapping nodes until it succeeds.
@@ -223,11 +222,10 @@ tryBootstrapJoining nsSTM = do
 bootstrapQueryId :: LocalNodeStateSTM s -> (String, PortNumber) -> NodeID -> IO (Either String RemoteNodeState)
 bootstrapQueryId nsSTM (bootstrapHost, bootstrapPort) targetID = do
     ns <- readTVarIO nsSTM
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns)
-    let srcAddr = confIP nodeConf
+    srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns)
     bootstrapResponse <- bracket (mkSendSocket srcAddr bootstrapHost bootstrapPort) close (
         -- Initialise an empty cache only with the responses from a bootstrapping node
-        fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (lookupMessage targetID ns Nothing)
+        fmap Right . sendRequestTo 5000 3 (lookupMessage targetID ns Nothing)
                                                                                   )
        `catch` (\e -> pure . Left $ "Error at bootstrap QueryId: " <> displayException (e :: IOException))
 
@@ -246,24 +244,26 @@ bootstrapQueryId nsSTM (bootstrapHost, bootstrapPort) targetID = do
                            Just (FORWARD resultset) -> foldr' (addCacheEntryPure now) cacheAcc resultset
                               )
                            initCache resp
-               currentlyResponsible <- runExceptT $ queryIdLookupLoop bootstrapCache ns 50 $ getNid ns
-               pure currentlyResponsible
+               currentlyResponsible <- queryIdLookupLoop bootstrapCache ns 50 $ getNid ns
+               pure $ Right currentlyResponsible
 
 
 -- | join a node to the DHT using the global node cache
 -- node's position.
-fediChordVserverJoin :: (MonadError String m, MonadIO m, Service s (RealNodeSTM s))
+fediChordVserverJoin :: Service s (RealNodeSTM s)
                      => LocalNodeStateSTM s                    -- ^ the local 'NodeState'
-                     -> m (LocalNodeStateSTM s)  -- ^ the joined 'NodeState' after a
+                     -> IO (Either String (LocalNodeStateSTM s))  -- ^ the joined 'NodeState' after a
                                                     -- successful join, otherwise an error message
 fediChordVserverJoin nsSTM = do
-    ns <- liftIO $ readTVarIO nsSTM
+    ns <- readTVarIO nsSTM
     -- 1. get routed to the currently responsible node
     currentlyResponsible <- requestQueryID ns $ getNid ns
-    liftIO . putStrLn $ "Trying to join on " <> show (getNid currentlyResponsible)
+    putStrLn $ "Trying to join on " <> show (getNid currentlyResponsible)
     -- 2. then send a join to the currently responsible node
-    joinResult <- liftIO $ requestJoin currentlyResponsible nsSTM
-    liftEither joinResult
+    joinResult <- requestJoin currentlyResponsible nsSTM
+    case joinResult of
+      Left err       -> pure . Left $ "Error joining on " <> err
+      Right joinedNS -> pure . Right $ joinedNS
 
 fediChordVserverLeave :: (MonadError String m, MonadIO m, Service s (RealNodeSTM s)) => LocalNodeState s -> m ()
 fediChordVserverLeave ns = do
@@ -311,23 +311,23 @@ joinOnNewEntriesThread nsSTM = loop
   where
     loop = do
         nsSnap <- readTVarIO nsSTM
-        (lookupResult, parentNode) <- atomically $ do
+        (lookupResult, cache) <- atomically $ do
             cache <- readTVar $ nodeCacheSTM nsSnap
-            parentNode <- readTVar $ parentRealNode nsSnap
             case queryLocalCache nsSnap cache 1 (getNid nsSnap) of
               -- empty cache, block until cache changes and then retry
               (FORWARD s) | Set.null s -> retry
-              result                   -> pure (result, parentNode)
+              result                   -> pure (result, cache)
         case lookupResult of
           -- already joined
           FOUND _ ->
               pure ()
           -- otherwise try joining
           FORWARD _ -> do
-              joinResult <- runExceptT $ fediChordVserverJoin nsSTM
+              joinResult <- fediChordVserverJoin nsSTM
               either
                 -- on join failure, sleep and retry
-                (const $ threadDelay (confJoinAttemptsInterval . nodeConfig $ parentNode) >> loop)
+                -- TODO: make delay configurable
+                (const $ threadDelay (30 * 10^6) >> loop)
                 (const $ pure ())
                 joinResult
 
@@ -342,15 +342,20 @@ nodeCacheWriter nsSTM =
         modifyTVar' (nodeCacheSTM ns) cacheModifier
 
 
+-- TODO: make max entry age configurable
+maxEntryAge :: POSIXTime
+maxEntryAge = 600
+
+
 -- | Periodically iterate through cache, clean up expired entries and verify unverified ones
 nodeCacheVerifyThread :: LocalNodeStateSTM s -> IO ()
 nodeCacheVerifyThread nsSTM = forever $ do
+    putStrLn "cache verify run: begin"
     -- get cache
-    (ns, cache, maxEntryAge) <- atomically $ do
+    (ns, cache) <- atomically $ do
         ns <- readTVar nsSTM
         cache <- readTVar $ nodeCacheSTM ns
-        maxEntryAge <- confMaxNodeCacheAge . nodeConfig <$> readTVar (parentRealNode ns)
-        pure (ns, cache, maxEntryAge)
+        pure (ns, cache)
     -- iterate entries:
     -- for avoiding too many time syscalls, get current time before iterating.
     now <- getPOSIXTime
@@ -397,7 +402,8 @@ nodeCacheVerifyThread nsSTM = forever $ do
         forkIO $ sendQueryIdMessages targetID latestNs (Just (1 + jEntriesPerSlice latestNs)) (nodesToQuery targetID) >> pure () -- ask for 1 entry more than j because of querying the middle
                                                              )
 
-    threadDelay $ fromEnum (maxEntryAge / 20) `div` 10^6    -- convert from pico to milliseconds
+    putStrLn "cache verify run: end"
+    threadDelay $ 10^6 * round maxEntryAge `div` 20
 
 
 -- | Checks the invariant of at least @jEntries@ per cache slice.
@@ -463,6 +469,7 @@ stabiliseThread :: Service s (RealNodeSTM s) => LocalNodeStateSTM s -> IO ()
 stabiliseThread nsSTM = forever $ do
     oldNs <- readTVarIO nsSTM
 
+    putStrLn "stabilise run: begin"
 
     -- iterate through the same snapshot, collect potential new neighbours
     -- and nodes to be deleted, and modify these changes only at the end of
@@ -497,26 +504,18 @@ stabiliseThread nsSTM = forever $ do
     -- try looking up additional neighbours if list too short
     forM_ [(length $ predecessors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
         ns' <- readTVarIO nsSTM
-        nextEntry <- runExceptT . requestQueryID ns' $ pred . getNid $ lastDef (toRemoteNodeState ns') (predecessors ns')
-        either
-            (const $ pure ())
-            (\entry -> atomically $ do
-                latestNs <- readTVar nsSTM
-                writeTVar nsSTM $ addPredecessors [entry] latestNs
-            )
-            nextEntry
+        nextEntry <- requestQueryID ns' $ pred . getNid $ lastDef (toRemoteNodeState ns') (predecessors ns')
+        atomically $ do
+            latestNs <- readTVar nsSTM
+            writeTVar nsSTM $ addPredecessors [nextEntry] latestNs
                                                                        )
 
     forM_ [(length $ successors updatedNs)..(kNeighbours updatedNs)] (\_ -> do
         ns' <- readTVarIO nsSTM
-        nextEntry <- runExceptT . requestQueryID ns' $ succ . getNid $ lastDef (toRemoteNodeState ns') (successors ns')
-        either
-            (const $ pure ())
-            (\entry -> atomically $ do
-                latestNs <- readTVar nsSTM
-                writeTVar nsSTM $ addSuccessors [entry] latestNs
-            )
-            nextEntry
+        nextEntry <- requestQueryID ns' $ succ . getNid $ lastDef (toRemoteNodeState ns') (successors ns')
+        atomically $ do
+            latestNs <- readTVar nsSTM
+            writeTVar nsSTM $ addSuccessors [nextEntry] latestNs
                                                                      )
 
     newNs <- readTVarIO nsSTM
@@ -541,8 +540,9 @@ stabiliseThread nsSTM = forever $ do
                     )
                     newPredecessor
 
-    stabiliseDelay <- confStabiliseInterval . nodeConfig <$> readTVarIO (parentRealNode newNs)
-    threadDelay stabiliseDelay
+    putStrLn "stabilise run: end"
+    -- TODO: make delay configurable
+    threadDelay (60 * 10^6)
   where
     -- | send a stabilise request to the n-th neighbour
     -- (specified by the provided getter function) and on failure retry
@@ -629,15 +629,19 @@ type RequestMap = Map.Map (SockAddr, Integer) RequestMapEntry
 data RequestMapEntry = RequestMapEntry (Set.Set FediChordMessage) (Maybe Integer)
                   POSIXTime
 
+-- TODO: make purge age configurable
+-- | periodically clean up old request parts
+responsePurgeAge :: POSIXTime
+responsePurgeAge = 60 -- seconds
 
-requestMapPurge :: POSIXTime -> MVar RequestMap -> IO ()
-requestMapPurge purgeAge mapVar = forever $ do
+requestMapPurge :: MVar RequestMap -> IO ()
+requestMapPurge mapVar = forever $ do
     rMapState <- takeMVar mapVar
     now <- getPOSIXTime
-    putMVar mapVar $ Map.filter (\(RequestMapEntry _ _ ts)  ->
-        now - ts < purgeAge
+    putMVar mapVar $ Map.filter (\entry@(RequestMapEntry _ _ ts)  ->
+        now - ts < responsePurgeAge
                                 ) rMapState
-    threadDelay $ (fromEnum purgeAge * 2) `div` 10^6
+    threadDelay $ round responsePurgeAge * 2 * 10^6
 
 
 -- | Wait for messages, deserialise them, manage parts and acknowledgement status,
@@ -652,13 +656,12 @@ fediMessageHandler sendQ recvQ nsSTM = do
     -- not change.
     -- Other functions are passed the nsSTM reference and thus can get the latest state.
     nsSnap <- readTVarIO nsSTM
-    nodeConf <- nodeConfig <$> readTVarIO (parentRealNode nsSnap)
     -- handling multipart messages:
     -- Request parts can be insert into a map (key: (sender IP against spoofing, request ID), value: timestamp + set of message parts, handle all of them when size of set == parts) before being handled. This map needs to be purged periodically by a separate thread and can be protected by an MVar for fairness.
     requestMap <- newMVar (Map.empty :: RequestMap)
     -- run receive loop and requestMapPurge concurrently, so that an exception makes
     -- both of them fail
-    concurrently_ (requestMapPurge (confResponsePurgeAge nodeConf) requestMap) $ forever $ do
+    concurrently_ (requestMapPurge requestMap) $ forever $ do
         -- wait for incoming messages
         (rawMsg, sourceAddr) <- atomically $ readTQueue recvQ
         let aMsg = deserialiseMessage rawMsg
@@ -713,7 +716,7 @@ fediMessageHandler sendQ recvQ nsSTM = do
 
 instance DHT (RealNodeSTM s) where
     lookupKey nodeSTM keystring = getKeyResponsibility nodeSTM $ genKeyID keystring
-    forceLookupKey nodeSTM keystring = (putStrLn $ "forced responsibility lookup of #" <> keystring) >> (updateLookupCache nodeSTM $ genKeyID keystring)
+    forceLookupKey nodeSTM keystring = updateLookupCache nodeSTM $ genKeyID keystring
     -- potential better implementation: put all neighbours of all vservers and the vservers on a ringMap, look the key up and see whether it results in a LocalNodeState
     isResponsibleFor nodeSTM key = do
         node <- readTVarIO nodeSTM
@@ -754,7 +757,7 @@ getKeyResponsibility nodeSTM lookupKey = do
 -- new entry.
 -- If no vserver is active in the DHT, 'Nothing' is returned.
 updateLookupCache :: RealNodeSTM s -> NodeID -> IO (Maybe (String, PortNumber))
-updateLookupCache nodeSTM keyToLookup = do
+updateLookupCache nodeSTM lookupKey = do
     (node, lookupSource) <- atomically $ do
         node <- readTVar nodeSTM
         let firstVs = headMay (vservers node)
@@ -764,25 +767,18 @@ updateLookupCache nodeSTM keyToLookup = do
         pure (node, lookupSource)
     maybe (do
         -- if no local node available, delete cache entry and return Nothing
-        atomically $ modifyTVar' (lookupCacheSTM node) $ Map.delete keyToLookup
+        atomically $ modifyTVar' (lookupCacheSTM node) $ Map.delete lookupKey
         pure Nothing
           )
           (\n -> do
         -- start a lookup from the node, update the cache with the lookup result and return it
-        -- TODO: better retry management, because having no vserver joined yet should
-        -- be treated differently than other reasons for not getting a result.
-        newResponsible <- runExceptT $ requestQueryID n keyToLookup
-        either
-            (const $ pure Nothing)
-            (\result -> do
-                let newEntry = (getDomain result, getServicePort result)
-                now <- getPOSIXTime
-                -- atomic update against lost updates
-                atomically $ modifyTVar' (lookupCacheSTM node) $
-                    Map.insert keyToLookup (CacheEntry False newEntry now)
-                pure $ Just newEntry
-            )
-            newResponsible
+        newResponsible <- requestQueryID n lookupKey
+        let newEntry = (getDomain newResponsible, getServicePort newResponsible)
+        now <- getPOSIXTime
+        -- atomic update against lost updates
+        atomically $ modifyTVar' (lookupCacheSTM node) $
+            Map.insert lookupKey (CacheEntry False newEntry now)
+        pure $ Just newEntry
            ) lookupSource
 
 
@@ -797,4 +793,4 @@ lookupCacheCleanup nodeSTM = do
                 now - ts < confMaxLookupCacheAge (nodeConfig node)
                        )
                                                        )
-        threadDelay $ fromEnum (2 * confMaxLookupCacheAge (nodeConfig node)) `div` 10^6
+        threadDelay $ round (confMaxLookupCacheAge $ nodeConfig node) * (10^5)

src/Hash2Pub/FediChordTypes.hs

@@ -411,22 +411,10 @@ data FediChordConf = FediChordConf
     -- ^ listening port for the FediChord DHT
     , confBootstrapNodes            :: [(String, PortNumber)]
     -- ^ list of potential bootstrapping nodes
-    , confStabiliseInterval         :: Int
-    -- ^ pause between stabilise runs, in milliseconds
     , confBootstrapSamplingInterval :: Int
-    -- ^ pause between sampling the own ID through bootstrap nodes, in milliseconds
+    -- ^ pause between sampling the own ID through bootstrap nodes, in seconds
     , confMaxLookupCacheAge         :: POSIXTime
-    -- ^ maximum age of key lookup cache entries in seconds
-    , confJoinAttemptsInterval      :: Int
-    -- ^ interval between join attempts on newly learned nodes, in milliseconds
-    , confMaxNodeCacheAge           :: POSIXTime
-    -- ^ maximum age of entries in the node cache, in milliseconds
-    , confResponsePurgeAge          :: POSIXTime
-    -- ^ maximum age of message parts in response part cache, in seconds
-    , confRequestTimeout            :: Int
-    -- ^ how long to wait until response has arrived, in milliseconds
-    , confRequestRetries            :: Int
-    -- ^ how often re-sending a timed-out request can be retried
+    -- ^ maximum age of lookup cache entries in seconds
     }
     deriving (Show, Eq)
 
@@ -457,12 +445,6 @@ data ServiceConf = ServiceConf
     -- ^ listening port for service
     , confServiceHost            :: String
     -- ^ hostname of service
-    , confLogfilePath            :: String
-    -- ^ where to store the (measurement) log file
-    , confStatsEvalDelay         :: Int
-    -- ^ delay between statistic rate measurement samplings, in microseconds
-    , confSpeedupFactor          :: Int
-    -- While the speedup factor needs to be already included in all
     }
 
 class DHT d where

src/Hash2Pub/PostService.hs

@@ -10,43 +10,41 @@ module Hash2Pub.PostService where
 
 import           Control.Concurrent
 import           Control.Concurrent.Async
+import           Control.Concurrent.MVar
 import           Control.Concurrent.STM
-import           Control.Exception         (Exception (..), try)
-import           Control.Monad             (foldM, forM, forM_, forever, unless,
-                                            void, when)
-import           Control.Monad.IO.Class    (liftIO)
+import           Control.Concurrent.STM.TChan
+import           Control.Concurrent.STM.TChan
+import           Control.Concurrent.STM.TQueue
+import           Control.Concurrent.STM.TVar
+import           Control.Exception             (Exception (..), try)
+import           Control.Monad                 (foldM, forM, forM_, forever)
+import           Control.Monad.IO.Class        (liftIO)
+import           Control.Monad.STM
 import           Data.Bifunctor
-import qualified Data.ByteString.Lazy.UTF8 as BSUL
-import qualified Data.ByteString.UTF8      as BSU
-import qualified Data.DList                as D
-import           Data.Either               (lefts, rights)
-import qualified Data.HashMap.Strict       as HMap
-import qualified Data.HashSet              as HSet
-import           Data.Maybe                (fromJust, isJust)
-import           Data.String               (fromString)
-import           Data.Text.Lazy            (Text)
-import qualified Data.Text.Lazy            as Txt
-import qualified Data.Text.Lazy.IO         as TxtI
-import           Data.Text.Normalize       (NormalizationMode (NFC), normalize)
+import qualified Data.ByteString.Lazy.UTF8     as BSUL
+import qualified Data.ByteString.UTF8          as BSU
+import qualified Data.HashMap.Strict           as HMap
+import qualified Data.HashSet                  as HSet
+import           Data.Maybe                    (fromMaybe, isJust)
+import           Data.String                   (fromString)
+import qualified Data.Text.Lazy                as Txt
+import           Data.Text.Normalize           (NormalizationMode (NFC),
+                                                normalize)
 import           Data.Time.Clock.POSIX
-import           Data.Typeable             (Typeable)
-import qualified Network.HTTP.Client       as HTTP
-import qualified Network.HTTP.Types        as HTTPT
-import           System.IO
+import           Data.Typeable                 (Typeable)
+import qualified Network.HTTP.Client           as HTTP
+import qualified Network.HTTP.Types            as HTTPT
 import           System.Random
-import           Text.Read                 (readEither)
+import           Text.Read                     (readEither)
 
-import           Formatting                (fixed, format, int, (%))
-import qualified Network.Wai.Handler.Warp  as Warp
+import qualified Network.Wai.Handler.Warp      as Warp
 import           Servant
 import           Servant.Client
+import           Servant.Server
 
 import           Hash2Pub.FediChordTypes
-import           Hash2Pub.PostService.API
 import           Hash2Pub.RingMap
-import           Hash2Pub.Utils
 
-import           Debug.Trace
 
 data PostService d = PostService
     { serviceConf          :: ServiceConf
@@ -57,22 +55,19 @@ data PostService d = PostService
     -- ^ for each tag store the subscribers + their queue
     , ownSubscriptions     :: TVar (HMap.HashMap NodeID POSIXTime)
     -- ^ tags subscribed by the own node have an assigned lease time
+    , ownPosts             :: TVar (HSet.HashSet Txt.Text)
+    -- ^ just store the existence of posts for saving memory,
     , relayInQueue         :: TQueue (Hashtag, PostID, PostContent)
     -- ^ Queue for processing incoming posts of own instance asynchronously
     , postFetchQueue       :: TQueue PostID
-    -- ^ queue of posts to be fetched
     , migrationsInProgress :: TVar (HMap.HashMap NodeID (MVar ()))
     , httpMan              :: HTTP.Manager
-    , statsQueue           :: TQueue StatsEvent
-    , loadStats            :: TVar RelayStats
-    -- ^ current load stats, replaced periodically
-    , logFileHandle        :: Handle
     }
     deriving (Typeable)
 
-type Hashtag = Text
-type PostID = Text
-type PostContent = Text
+type Hashtag = Txt.Text
+type PostID = Txt.Text
+type PostContent = Txt.Text
 -- | For each handled tag, store its subscribers and provide a
 -- broadcast 'TChan' for enqueuing posts
 type RelayTags = RingMap NodeID (TagSubscribersSTM, TChan PostID, Hashtag)
@@ -90,38 +85,26 @@ instance DHT d => Service PostService d where
         threadVar <- newTVarIO =<< myThreadId -- own thread ID as placeholder
         subscriberVar <- newTVarIO emptyRMap
         ownSubsVar <- newTVarIO HMap.empty
-        --ownPostVar <- newTVarIO HSet.empty
+        ownPostVar <- newTVarIO HSet.empty
         relayInQueue' <- newTQueueIO
         postFetchQueue' <- newTQueueIO
         migrationsInProgress' <- newTVarIO HMap.empty
         httpMan' <- HTTP.newManager HTTP.defaultManagerSettings
-        statsQueue' <- newTQueueIO
-        loadStats' <- newTVarIO emptyStats
-        loggingFile <- openFile (confLogfilePath conf) WriteMode
-        hSetBuffering loggingFile LineBuffering
         let
-            thisService = PostService
-              { serviceConf = conf
+            thisService = PostService {
+                serviceConf = conf
               , baseDHT = dht
               , serviceThread = threadVar
               , subscribers = subscriberVar
               , ownSubscriptions = ownSubsVar
-              --, ownPosts = ownPostVar
+              , ownPosts = ownPostVar
               , relayInQueue = relayInQueue'
               , postFetchQueue = postFetchQueue'
               , migrationsInProgress = migrationsInProgress'
               , httpMan = httpMan'
-              , statsQueue = statsQueue'
-              , loadStats = loadStats'
-              , logFileHandle = loggingFile
-              }
+                                      }
             port' = fromIntegral (confServicePort conf)
             warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings
-        -- log a start message, this also truncates existing files
-        TxtI.hPutStrLn loggingFile $ Txt.unlines
-            [ "# Starting mock relay implementation"
-            , "#time stamp         ; relay receive rate   ;relay delivery rate   ;instance publish rate ;instance fetch rate   ;total subscriptions"
-            ]
         -- Run 'concurrently_' from another thread to be able to return the
         -- 'PostService'.
         -- Terminating that parent thread will make all child threads terminate as well.
@@ -129,11 +112,7 @@ instance DHT d => Service PostService d where
             concurrently_
                 -- web server
                 (Warp.runSettings warpSettings $ postServiceApplication thisService)
-                $ concurrently
-                    -- background processing workers
-                    (launchWorkerThreads thisService)
-                    -- statistics/ measurements
-                    (launchStatsThreads thisService)
+                (processIncomingPosts thisService)
         -- update thread ID after fork
         atomically $ writeTVar threadVar servThreadID
         pure thisService
@@ -159,13 +138,38 @@ postServiceApplication :: DHT d => PostService d -> Application
 postServiceApplication serv = serve exposedPostServiceAPI $ postServer serv
 
 
+-- | needed for guiding type inference
+exposedPostServiceAPI :: Proxy PostServiceAPI
+exposedPostServiceAPI = Proxy
+
 -- ========= constants ===========
 
-placeholderPost :: Text
+placeholderPost :: Txt.Text
 placeholderPost = Txt.take 5120 . Txt.repeat $ 'O' -- size 5KiB
 
 -- ========= HTTP API and handlers =============
 
+type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent
+                 -- delivery endpoint of newly published posts of the relay's instance
+                 :<|> "relay" :> "subscribers" :> Capture "senderID" Integer :> ReqBody '[PlainText] Txt.Text :> PostNoContent '[PlainText] Txt.Text
+                 -- endpoint for delivering the subscriptions and outstanding queue
+                 :<|> "post" :> Capture "postid" Txt.Text :> Get '[PlainText] Txt.Text
+                 -- fetch endpoint for posts, full post ID is http://$domain/post/$postid
+                 :<|> "posts" :> ReqBody '[PlainText] Txt.Text :> Post '[PlainText] Txt.Text
+                 -- endpoint for fetching multiple posts at once
+                 :<|> "posts" :> "inbox" :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent
+                 -- delivery endpoint of newly published posts of the relay's instance
+                 :<|> "tags" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PostCreated '[PlainText] Txt.Text
+                 -- delivery endpoint for posts of $tag at subscribing instance
+                 :<|> "tags" :> Capture "hashtag" Txt.Text :> "subscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Integer
+                 -- endpoint for subscribing the instance specified in
+                 -- the Origin header to $hashtag.
+                 -- Returns subscription lease time in seconds.
+                 :<|> "tags" :> Capture "hashtag" Txt.Text :> "unsubscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Txt.Text
+                 -- endpoint for unsubscribing the instance specified in
+                 -- the Origin header to $hashtag
+
+
 postServer :: DHT d => PostService d -> Server PostServiceAPI
 postServer service = relayInbox service
         :<|> subscriptionDelivery service
@@ -177,27 +181,24 @@ postServer service = relayInbox service
         :<|> tagUnsubscribe service
 
 
--- | delivery endpoint: receive posts of a handled tag and enqueue them for relaying
-relayInbox :: DHT d => PostService d -> Hashtag -> Text -> Handler NoContent
+relayInbox :: DHT d => PostService d -> Hashtag -> Txt.Text -> Handler NoContent
 relayInbox serv tag posts = do
     let
     -- skip checking whether the post actually contains the tag, just drop full post
         postIDs = head . Txt.splitOn "," <$> Txt.lines posts
     -- if tag is not in own responsibility, return a 410 Gone
-    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId tag)
+    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (genKeyID . Txt.unpack $ tag)
     if responsible
        then pure ()
        else
-        throwError $ err410 { errBody = "Relay is not responsible for this tag"}
+        (throwError $ err410 { errBody = "Relay is not responsible for this tag"})
     broadcastChan <- liftIO $ atomically $ getTagBroadcastChannel serv tag
     maybe
     -- if noone subscribed to the tag, nothing needs to be done
         (pure ())
     -- otherwise enqueue posts into broadcast queue of the tag
-        (\queue -> do
+        (\queue ->
             liftIO $ forM_ postIDs (atomically . writeTChan queue)
-            -- report the received post for statistic purposes
-            liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent RelayReceiveEvent (length postIDs) (hashtagToId tag)
         )
         broadcastChan
     pure NoContent
@@ -208,8 +209,7 @@ newtype UnhandledTagException = UnhandledTagException String
 
 instance Exception UnhandledTagException
 
--- | delivery endpoint: receives a list of subscribers of tags and their outstanding queues for migration
-subscriptionDelivery :: DHT d => PostService d -> Integer -> Text -> Handler Text
+subscriptionDelivery :: DHT d => PostService d -> Integer -> Txt.Text -> Handler Txt.Text
 subscriptionDelivery serv senderID subList = do
     let
         tagSubs = Txt.lines subList
@@ -221,7 +221,7 @@ subscriptionDelivery serv senderID subList = do
     -- not-handled tag occurs, this results in a single large transaction.
     -- Hopefully the performance isn't too bad.
     res <- liftIO . atomically $ (foldM (\_ tag' -> do
-        responsible <- isResponsibleForSTM (baseDHT serv) (hashtagToId tag')
+        responsible <- isResponsibleForSTM (baseDHT serv) (genKeyID . Txt.unpack $ tag')
         if responsible
            then processTag (subscribers serv) tag'
            else throwSTM $ UnhandledTagException (Txt.unpack tag' <> " not handled by this relay")
@@ -243,7 +243,7 @@ subscriptionDelivery serv senderID subList = do
       Right _  -> pure ""
     -- TODO: check and only accept tags in own (future?) responsibility
   where
-      processTag :: TVar RelayTags -> Text -> STM ()
+      processTag :: TVar RelayTags -> Txt.Text -> STM ()
       processTag subscriberSTM tagData = do
           let
             tag:subText:lease:posts:_ = Txt.splitOn "," tagData
@@ -254,62 +254,57 @@ subscriptionDelivery serv senderID subList = do
           enqueueSubscription subscriberSTM (normaliseTag tag) sub postList leaseTime
 
 
--- | endpoint for fetching a post by its ID
-postFetch :: PostService d -> Text -> Handler Text
-postFetch serv _ = do
-    -- decision: for saving memory do not store published posts, just
-    -- pretend there is a post for each requested ID
-    liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent 1 0  -- tag fetched for is irrelevant
-    pure placeholderPost
+postFetch :: PostService d -> Txt.Text -> Handler Txt.Text
+postFetch serv postID = do
+    postSet <- liftIO . readTVarIO . ownPosts $ serv
+    if HSet.member postID postSet
+       -- decision: always return the same placeholder post
+       then pure placeholderPost
+       else throwError $ err404 { errBody = "No post found with this ID" }
 
 
--- | endpoint for fetching multiple posts of this instance by their IDs
-postMultiFetch :: PostService d -> Text -> Handler Text
+postMultiFetch :: PostService d -> Txt.Text -> Handler Txt.Text
 postMultiFetch serv postIDs = do
-    let
-        idList = Txt.lines postIDs
-        -- decision: for saving memory do not store published posts, just
-        -- pretend there is a post for each requested ID
-        response = foldl (\response' _ ->
-           placeholderPost <> "\n" <> response'
+    let idList = Txt.lines postIDs
+    postSet <- liftIO . readTVarIO . ownPosts $ serv
+    -- look up existence of all given post IDs, fail if even one is missing
+    foldM (\response postID ->
+        if HSet.member postID postSet
+           then pure $ placeholderPost <> "\n" <> response
+           else throwError $ err404 { errBody = "No post found with this ID" }
           ) "" idList
-    liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent (length idList) 0  -- tag fetched for is irrelevant
-    pure response
 
 
--- | delivery endpoint: inbox for initially publishing a post at an instance
-postInbox :: PostService d -> Text -> Handler NoContent
+postInbox :: PostService d -> Txt.Text -> Handler NoContent
 postInbox serv post = do
     -- extract contained hashtags
     let
         containedTags = fmap (normaliseTag . Txt.tail) . filter ((==) '#' . Txt.head) . Txt.words $ post
     -- generate post ID
     postId <- liftIO $ Txt.pack . show <$> (randomRIO (0, 2^(128::Integer)-1) :: IO Integer)
-    -- decision: for saving memory do not store published post IDs, just deliver a post for any requested ID
+    -- add ID to own posts
+    liftIO . atomically $ modifyTVar' (ownPosts serv) (HSet.insert postId)
     -- enqueue a relay job for each tag
-    liftIO $ forM_ (containedTags :: [Text]) (\tag ->
+    liftIO $ forM_ (containedTags :: [Txt.Text]) (\tag ->
         atomically $ writeTQueue (relayInQueue serv) (tag, postId,  post)
                                  )
     pure NoContent
 
 
--- | delivery endpoint: receive postIDs of a certain subscribed hashtag
-tagDelivery :: PostService d -> Text -> Text -> Handler Text
+tagDelivery :: PostService d -> Txt.Text -> Txt.Text -> Handler Txt.Text
 tagDelivery serv hashtag posts = do
     let postIDs = Txt.lines posts
     subscriptions <- liftIO . readTVarIO . ownSubscriptions $ serv
-    if isJust (HMap.lookup (hashtagToId hashtag) subscriptions)
+    if isJust (HMap.lookup (genKeyID . Txt.unpack $ hashtag) subscriptions)
        then -- TODO: increase a counter/ statistics for received posts of this tag
             liftIO $ forM_ postIDs $ atomically . writeTQueue (postFetchQueue serv)
         else -- silently drop posts from unsubscribed tags
             pure ()
     pure $ "Received a postID for tag " <> hashtag
 
-
--- | receive subscription requests to a handled hashtag
-tagSubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Integer
+tagSubscribe :: DHT d => PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Integer
 tagSubscribe serv hashtag origin = do
-    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag)
+    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (genKeyID . Txt.unpack $ hashtag)
     if not responsible
        -- GONE if not responsible
        then throwError err410 { errBody = "not responsible for this tag" }
@@ -323,14 +318,12 @@ tagSubscribe serv hashtag origin = do
     let leaseTime = now + confSubscriptionExpiryTime (serviceConf serv)
     -- setup subscription entry
     _ <- liftIO . atomically $ setupSubscriberChannel (subscribers serv) hashtag (BSU.toString $ HTTP.host req, HTTP.port req) leaseTime
-    --liftIO . putStrLn $ "just got a subscription to " <> Txt.unpack hashtag
     pure $ round leaseTime
 
 
--- | receive and handle unsubscription requests regarding a handled tag
-tagUnsubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Text
+tagUnsubscribe :: DHT d => PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Txt.Text
 tagUnsubscribe serv hashtag origin = do
-    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag)
+    responsible <- liftIO $ isResponsibleFor (baseDHT serv) (genKeyID . Txt.unpack $ hashtag)
     if not responsible
        -- GONE if not responsible
        then throwError err410 { errBody = "not responsible for this tag" }
@@ -348,15 +341,8 @@ tagUnsubscribe serv hashtag origin = do
 clientAPI :: Proxy PostServiceAPI
 clientAPI = Proxy
 
-relayInboxClient
-    :<|> subscriptionDeliveryClient
-    :<|> postFetchClient
-    :<|> postMultiFetchClient
-    :<|> postInboxClient
-    :<|> tagDeliveryClient
-    :<|> tagSubscribeClient
-    :<|> tagUnsubscribeClient
-    = client clientAPI
+
+relayInboxClient :<|> subscriptionDeliveryClient :<|> postFetchClient :<|> postMultiFetchClient :<|> postInboxClient :<|> tagDeliveryClient :<|> tagSubscribeClient :<|> tagUnsubscribeClient = client clientAPI
 
 
 -- | Deliver the subscriber list of all hashtags in the interval [fromTag, toTag]
@@ -369,7 +355,7 @@ clientDeliverSubscriptions :: PostService d
                            -> (String, Int)     -- ^ hostname and port of instance to deliver to
                            -> IO (Either String ())     -- Either signals success or failure
 clientDeliverSubscriptions serv fromNode fromKey toKey (toHost, toPort) = do
-    -- collect tag interval
+    -- collect tag intearval
     intervalTags <- takeRMapSuccessorsFromTo fromKey toKey <$> readTVarIO (subscribers serv)
     -- returns a [ (TagSubscribersSTM, TChan PostID, Hashtag) ]
     -- extract subscribers and posts
@@ -399,7 +385,7 @@ clientDeliverSubscriptions serv fromNode fromKey toKey (toHost, toPort) = do
       Right _ -> do
           atomically $
               modifyTVar' (subscribers serv) $ \tagMap ->
-                  foldr deleteRMapEntry tagMap ((\(_, _, t) -> hashtagToId t) <$> intervalTags)
+                  foldr deleteRMapEntry tagMap ((\(_, _, t) -> genKeyID . Txt.unpack $ t) <$> intervalTags)
           pure . Right $ ()
   where
       channelGetAll :: TChan a -> STM [a]
@@ -410,8 +396,7 @@ clientDeliverSubscriptions serv fromNode fromKey toKey (toHost, toPort) = do
           maybe (pure acc) (\x -> channelGetAll' chan (x:acc)) haveRead
 
 
--- | Subscribe the client to the given hashtag. On success it returns the given lease time,
--- but also records the subscription in its own data structure.
+-- | Subscribe the client to the given hashtag. On success it returns the given lease time.
 clientSubscribeTo :: DHT d => PostService d -> Hashtag -> IO (Either String Integer)
 clientSubscribeTo serv tag = do
     lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag)
@@ -426,13 +411,9 @@ clientSubscribeTo serv tag = do
                 Left (FailureResponse _ fresp)
                   |(HTTPT.statusCode . responseStatusCode $ fresp) == 410 && allowRetry -> do  -- responsibility gone, force new lookup
                       newRes <- forceLookupKey (baseDHT serv) (Txt.unpack tag)
-                      --putStrLn $ "failed subscribing to " <> Txt.unpack tag <> " on " <> foundHost
                       doSubscribe newRes False
                 Left err    -> pure . Left . show $ err
-                Right lease -> do
-                    atomically . modifyTVar' (ownSubscriptions serv) $ HMap.insert (hashtagToId tag) (fromInteger lease)
-                    --putStrLn $ "just subscribed to " <> Txt.unpack tag <> " on " <> foundHost
-                    pure . Right $ lease
+                Right lease -> pure . Right $ lease
           )
           lookupResponse
 
@@ -454,9 +435,7 @@ clientUnsubscribeFrom serv tag = do
                       newRes <- forceLookupKey (baseDHT serv) (Txt.unpack tag)
                       doUnsubscribe newRes False
                 Left err    -> pure . Left . show $ err
-                Right _ -> do
-                    atomically . modifyTVar' (ownSubscriptions serv) $ HMap.delete (hashtagToId tag)
-                    pure . Right $ ()
+                Right _ -> pure . Right $ ()
           )
           lookupResponse
 
@@ -466,11 +445,11 @@ clientUnsubscribeFrom serv tag = do
 -- the post to the responsible relays.
 -- As the initial publishing isn't done by a specific relay (but *to* a specific relay
 -- instead), the function does *not* take a PostService as argument.
-clientPublishPost :: HTTP.Manager   -- ^ for better performance, a shared HTTP manager has to be provided
-                  -> String         -- ^ hostname
-                  -> Int            -- ^ port
-                  -> PostContent    -- ^ post content
-                  -> IO (Either String ())  -- ^ error or success
+clientPublishPost :: HTTP.Manager   -- for better performance, a shared HTTP manager has to be provided
+                  -> String         -- hostname
+                  -> Int            -- port
+                  -> PostContent    -- post content
+                  -> IO (Either String ())  -- error or success
 clientPublishPost httpman hostname port postC = do
     resp <- runClientM (postInboxClient postC) (mkClientEnv httpman (BaseUrl Http hostname port ""))
     pure . bimap show (const ()) $ resp
@@ -513,7 +492,7 @@ setupSubscriberChannel tagMapSTM tag subscriber leaseTime = do
         broadcastChan <- newBroadcastTChan
         tagOutChan <- dupTChan broadcastChan
         newSubMapSTM <- newTVar $ HMap.singleton subscriber (tagOutChan, leaseTime)
-        writeTVar tagMapSTM $ addRMapEntry (hashtagToId tag) (newSubMapSTM, broadcastChan, tag) tagMap
+        writeTVar tagMapSTM $ addRMapEntry (genKeyID . Txt.unpack $ tag) (newSubMapSTM, broadcastChan, tag) tagMap
         pure tagOutChan
       Just (foundSubMapSTM, broadcastChan, _) -> do
         -- otherwise use the existing subscriber map
@@ -541,7 +520,7 @@ deleteSubscription tagMapSTM tag subscriber = do
           -- if there are no subscriptions for the tag anymore, remove its
           -- data sttructure altogether
           if HMap.null newSubMap
-             then writeTVar tagMapSTM $ deleteRMapEntry (hashtagToId tag) tagMap
+             then writeTVar tagMapSTM $ deleteRMapEntry (genKeyID . Txt.unpack $ tag) tagMap
           -- otherwise just remove the subscription of that node
              else writeTVar foundSubMapSTM newSubMap
 
@@ -562,39 +541,12 @@ getTagBroadcastChannel serv tag = do
 
 -- | look up the subscription data of a tag
 lookupTagSubscriptions :: Hashtag -> RingMap NodeID a -> Maybe a
-lookupTagSubscriptions tag = rMapLookup (hashtagToId tag)
+lookupTagSubscriptions tag = rMapLookup (genKeyID . Txt.unpack $ tag)
 
 
--- normalise the unicode representation of a string to NFC and convert to lower case
-normaliseTag :: Text -> Text
-normaliseTag = Txt.toLower . Txt.fromStrict . normalize NFC . Txt.toStrict
-
-
--- | convert a hashtag to its representation on the DHT
-hashtagToId :: Hashtag -> NodeID
-hashtagToId = genKeyID . Txt.unpack
-
-
-readUpToTChan :: Int -> TChan a -> STM [a]
-readUpToTChan 0 _ = pure []
-readUpToTChan n chan = do
-  readFromChan <- tryReadTChan chan
-  case readFromChan of
-       Nothing -> pure []
-       Just val -> do
-           moreReads <- readUpToTChan (pred n) chan
-           pure (val:moreReads)
-
-
-readUpToTQueue :: Int -> TQueue a -> STM [a]
-readUpToTQueue 0 _ = pure []
-readUpToTQueue n q = do
-    readFromQueue <- tryReadTQueue q
-    case readFromQueue of
-      Nothing -> pure []
-      Just val -> do
-          moreReads <- readUpToTQueue (pred n) q
-          pure (val:moreReads)
+-- normalise the unicode representation of a string to NFC
+normaliseTag :: Txt.Text -> Txt.Text
+normaliseTag = Txt.fromStrict . normalize NFC . Txt.toStrict
 
 -- | define how to convert all showable types to PlainText
 -- No idea what I'm doing with these overlappable instances though ¯\_(ツ)_/¯
@@ -607,78 +559,28 @@ instance {-# OVERLAPPABLE #-} Read a => MimeUnrender PlainText a where
 
 -- ====== worker threads ======
 
--- TODO: make configurable
-numParallelDeliveries = 10
-
-launchWorkerThreads :: DHT d => PostService d -> IO ()
-launchWorkerThreads serv = concurrently_
-    (processIncomingPosts serv)
-    $ concurrently_
-        (purgeSubscriptionsThread serv)
-        $ concurrently_
-            (fetchTagPosts serv)
-            (relayWorker serv)
-
-
--- | periodically remove expired subscription entries from relay subscribers
-purgeSubscriptionsThread :: PostService d -> IO ()
-purgeSubscriptionsThread serv = forever $ do
-    -- read config
-    now <- getPOSIXTime
-    let
-        purgeInterval = confSubscriptionExpiryTime (serviceConf serv) / 10
-    -- no need to atomically lock this, as newly incoming subscriptions do not
-    -- need to be purged
-    tagMap <- readTVarIO $ subscribers serv
-    forM_ tagMap $ \(subscriberMapSTM, _, _) ->
-        -- but each subscriberMap needs to be modified atomically
-        atomically . modifyTVar' subscriberMapSTM $ HMap.filter (\(_, ts) -> ts > now)
-    threadDelay $ fromEnum purgeInterval `div` 10^6
-
-
 -- | process the pending relay inbox of incoming posts from the internal queue:
 -- Look up responsible relay node for given hashtag and forward post to it
 processIncomingPosts :: DHT d => PostService d -> IO ()
 processIncomingPosts serv = forever $ do
     -- blocks until available
-    deliveriesToProcess <- atomically $ do
-        readResult <- readUpToTQueue numParallelDeliveries $ relayInQueue serv
-        if null readResult
-           then retry
-           else pure readResult
-    runningJobs <- forM deliveriesToProcess $ \(tag, pID, pContent) -> async $ do
-        let pIdUri = "http://" <> (Txt.pack . confServiceHost . serviceConf $ serv) <> ":" <> (fromString . show . confServicePort . serviceConf $ serv) <> "/post/" <> pID
-        lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag)
-        case lookupRes of
-          -- no vserver active => wait and retry
-          Nothing -> threadDelay (10 * 10^6) >> pure (Left "no vserver active")
-          Just (responsibleHost, responsiblePort) -> do
-            resp <- runClientM (relayInboxClient tag $ pIdUri <> "," <> pContent) (mkClientEnv (httpMan serv) (BaseUrl Http responsibleHost (fromIntegral responsiblePort) ""))
-            case resp of
-              Left err  -> do
-                  -- 410 error indicates outdated responsibility mapping
-                  -- Simplification: just invalidate the mapping entry on all errors, force a re-lookup and re-queue the post
-                  -- TODO: keep track of maximum retries
-                  _ <- forceLookupKey (baseDHT serv) (Txt.unpack tag)
-                  atomically . writeTQueue (relayInQueue serv) $ (tag, pID, pContent)
-                  pure . Left $ "Error: " <> show err
-              Right _ -> do
-                  -- idea for the experiment: each post publication makes the initial posting instance subscribe to all contained tags
-                  now <- getPOSIXTime
-                  subscriptionStatus <- HMap.lookup (hashtagToId tag) <$> readTVarIO (ownSubscriptions serv)
-                  -- if not yet subscribed or subscription expires within 5 minutes, (re)subscribe to tag
-                  when (maybe True (\subLease -> now - subLease < 300) subscriptionStatus) $
-                      void $ clientSubscribeTo serv tag
-
-                  -- for evaluation, return the tag of the successfully forwarded post
-                  pure $ Right tag
-
-    -- collect async results
-    results <- mapM waitCatch runningJobs
-    -- report the count of published posts for statistics
-    atomically . writeTQueue (statsQueue serv) $ StatsEvent PostPublishEvent (length . rights $ results) 0    -- hashtag published to doesn't matter
-    pure ()
-
+    -- TODO: process multiple in parallel
+    (tag, pID, pContent) <- atomically . readTQueue $ relayInQueue serv
+    lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag)
+    case lookupRes of
+      -- no vserver active => wait and retry
+      Nothing -> threadDelay $ 10 * 10^6
+      Just (responsibleHost, responsiblePort) -> do
+        resp <- runClientM (relayInboxClient tag $ pID <> "," <> pContent) (mkClientEnv (httpMan serv) (BaseUrl Http responsibleHost (fromIntegral responsiblePort) ""))
+        case resp of
+          Left err  -> do
+              putStrLn $ "Error: " <> show err
+              -- 410 error indicates outdated responsibility mapping
+              -- Simplification: just invalidate the mapping entry on all errors, force a re-lookup and re-queue the post
+              -- TODO: keep track of maximum retries
+              _ <- forceLookupKey (baseDHT serv) (Txt.unpack tag)
+              atomically . writeTQueue (relayInQueue serv) $ (tag, pID, pContent)
+          Right yay -> putStrLn $ "Yay! " <> show yay
 
 
 -- | process the pending fetch jobs of delivered post IDs: Delivered posts are tried to be fetched from their URI-ID
@@ -688,197 +590,18 @@ fetchTagPosts serv = forever $ do
     -- TODO: batching, retry
     -- TODO: process multiple in parallel
     pIdUri <- atomically . readTQueue $ postFetchQueue serv
-    fetchReq <- HTTP.parseRequest . Txt.unpack $ pIdUri
+    fetchReq <- HTTP.parseRequest . Txt.unpack $pIdUri
     resp <- try $ HTTP.httpLbs fetchReq (httpMan serv) :: IO (Either HTTP.HttpException (HTTP.Response BSUL.ByteString))
     case resp of
       Right response ->
-          -- TODO error handling, retry
-          --if HTTPT.statusCode (HTTP.responseStatus response) == 200
-          --   then
-          --   -- success, TODO: statistics
-          --   else
-          pure ()
+          if HTTPT.statusCode (HTTP.responseStatus response) == 200
+             then
+             -- success, TODO: statistics
+             putStrLn "post fetch success"
+             else
+             -- TODO error handling, retry
+             pure ()
       Left _ ->
              -- TODO error handling, retry
              pure ()
 
-
-relayWorker :: PostService d -> IO ()
-relayWorker serv = forever $ do
-    -- atomically (to be able to retry) fold a list of due delivery actions
-    jobsToProcess <- atomically $ do
-        subscriptionMap <- readTVar $ subscribers serv
-        jobList <- D.toList <$> foldM (\jobAcc (subscriberMapSTM, _, tag) -> do
-            subscriberMap <- readTVar subscriberMapSTM
-            foldM (\jobAcc' ((subHost, subPort), (postChan, _)) -> do
-                postsToDeliver <- readUpToTChan 500 postChan
-                let postDeliveryAction = runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) ""))
-                -- append relay push job to job list
-                pure $ if not (null postsToDeliver)
-                          then jobAcc' `D.snoc` (do
-                              deliveryResult <- postDeliveryAction
-                              either
-                                (const $ pure ())
-                                -- on successful push, record that event for statistics
-                                (const . atomically . writeTQueue (statsQueue serv) $ StatsEvent RelayDeliveryEvent (length postsToDeliver) (hashtagToId tag))
-                                deliveryResult
-                              pure deliveryResult
-                                                )
-                   else jobAcc'
-                ) jobAcc $ HMap.toList subscriberMap
-            ) D.empty subscriptionMap
-        -- if no relay jobs, then retry
-        if null jobList
-           then retry
-           else pure jobList
-
-    -- when processing the list, send several deliveries in parallel
-    forM_ (chunksOf numParallelDeliveries jobsToProcess) $ \jobset -> do
-        runningJobs <- mapM async jobset
-        -- so far just dropping failed attempts, TODO: retry mechanism
-        results <- mapM waitCatch runningJobs
-        let
-            successfulResults =  rights results
-            unsuccessfulResults = lefts results
-        unless (null unsuccessfulResults) $ putStrLn ("ERR: " <> show (length unsuccessfulResults) <> " failed deliveries!")
-        putStrLn $ "successfully relayed " <> show (length successfulResults)
-        pure ()
-
-
--- ======= statistics/measurement and logging =======
-
-data StatsEventType = PostPublishEvent
-    | RelayReceiveEvent
-    | RelayDeliveryEvent
-    | IncomingPostFetchEvent
-    deriving (Enum, Show, Eq)
-
--- | Represents measurement event of a 'StatsEventType' with a count relevant for a certain key
-data StatsEvent = StatsEvent StatsEventType Int NodeID
-    deriving (Show, Eq)
-
-
--- | measured rates of relay performance
--- TODO: maybe include other metrics in here as well, like number of subscribers?
-data RelayStats = RelayStats
-    { relayReceiveRates  :: RingMap NodeID Double
-    -- ^ rate of incoming posts in the responsibility of this relay
-    , relayDeliveryRates :: RingMap NodeID Double
-    -- ^ rate of relayed outgoing posts
-    , postFetchRate      :: Double -- no need to differentiate between tags
-    -- ^ number of post-fetches delivered
-    , postPublishRate    :: Double
-    -- ^ rate of initially publishing posts through this instance
-    }
-    deriving (Show, Eq)
-
-
-
-launchStatsThreads :: PostService d -> IO ()
-launchStatsThreads serv = do
-    -- create shared accumulator
-    sharedAccum <- newTVarIO emptyStats
-    concurrently_
-        (accumulateStatsThread sharedAccum $ statsQueue serv)
-        (evaluateStatsThread serv sharedAccum)
-
-
--- | Read stats events from queue and add them to a shared accumulator.
--- Instead of letting the events accumulate in the queue and allocate linear memory, immediately fold the result.
-accumulateStatsThread :: TVar RelayStats -> TQueue StatsEvent -> IO ()
-accumulateStatsThread statsAccumulator statsQ = forever $ do
-    -- blocks until stats event arrives
-    event <- atomically $ readTQueue statsQ
-    -- add the event number to current accumulator
-    atomically $ modifyTVar' statsAccumulator $ statsAdder event
-
-
--- | add incoming stats events to accumulator value
-statsAdder :: StatsEvent -> RelayStats -> RelayStats
-statsAdder event stats = case event of
-    StatsEvent PostPublishEvent num _ ->
-        stats {postPublishRate = fromIntegral num + postPublishRate stats}
-    StatsEvent RelayReceiveEvent num key ->
-        stats {relayReceiveRates = sumIfEntryExists key (fromIntegral num) (relayReceiveRates stats)}
-    StatsEvent RelayDeliveryEvent num key ->
-        stats {relayDeliveryRates = sumIfEntryExists key (fromIntegral num) (relayDeliveryRates stats)}
-    StatsEvent IncomingPostFetchEvent num _ ->
-        stats {postFetchRate = fromIntegral num + postFetchRate stats}
-  where
-      sumIfEntryExists = addRMapEntryWith (\newVal oldVal ->
-          let toInsert = fromJust $ extractRingEntry newVal
-          in
-          case oldVal of
-            KeyEntry n -> KeyEntry (n + toInsert)
-            ProxyEntry pointer (Just (KeyEntry n)) -> ProxyEntry pointer (Just (KeyEntry $ n + toInsert))
-            ProxyEntry pointer Nothing -> ProxyEntry pointer (Just newVal)
-            _ -> error "RingMap nested too deeply"
-                                          )
-
-
--- Periodically exchange the accumulated statistics with empty ones, evaluate them
--- and make them the current statistics of the service.
-evaluateStatsThread :: PostService d -> TVar RelayStats -> IO ()
-evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop
-  where
-      loop previousTs = do
-          threadDelay $ confStatsEvalDelay (serviceConf serv)
-          -- get and reset the stats accumulator
-          summedStats <- atomically $ do
-              stats <- readTVar statsAcc
-              writeTVar statsAcc emptyStats
-              pure stats
-          -- as the transaction might retry several times, current time needs to
-          -- be read afterwards
-          now <- getPOSIXTime
-          -- evaluate stats rate and replace server stats
-          -- persistently store in a TVar so it can be retrieved later by the DHT
-          let timePassed = (now - previousTs) * fromIntegral (confSpeedupFactor $ serviceConf serv)
-              rateStats = evaluateStats timePassed summedStats
-          atomically $ writeTVar (loadStats serv) rateStats
-          -- and now what? write a log to file
-          -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate; subscriberSum
-          -- later: current (reported) load, target load
-          subscriberSum <- sumSubscribers
-          TxtI.hPutStrLn (logFileHandle serv) $
-              format (fixed 9 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % int )
-                  (realToFrac now :: Double)
-                  (sum . relayReceiveRates $ rateStats)
-                  (sum . relayDeliveryRates $ rateStats)
-                  (postPublishRate rateStats)
-                  (postFetchRate rateStats)
-                  subscriberSum
-          loop now
-
-      sumSubscribers = do
-          tagMap <- readTVarIO $ subscribers serv
-          foldM (\subscriberSum (subscriberMapSTM, _, _) -> do
-              subscriberMap <- readTVarIO subscriberMapSTM
-              pure $ subscriberSum + HMap.size subscriberMap
-                )
-                0 tagMap
-
-
--- | Evaluate the accumulated statistic events: Currently mostly calculates the event
--- rates by dividing through the collection time frame
-evaluateStats :: POSIXTime -> RelayStats -> RelayStats
-evaluateStats timeInterval summedStats =
-    -- first sum all event numbers, then divide through number of seconds passed to
-    -- get rate per second
-    RelayStats
-    { relayReceiveRates = (/ intervalSeconds) <$> relayReceiveRates summedStats
-    , relayDeliveryRates = (/ intervalSeconds) <$> relayDeliveryRates summedStats
-    , postPublishRate = postPublishRate summedStats / intervalSeconds
-    , postFetchRate = postFetchRate summedStats / intervalSeconds
-    }
-  where
-      intervalSeconds = realToFrac timeInterval
-
-
-emptyStats :: RelayStats
-emptyStats = RelayStats
-    { relayReceiveRates = emptyRMap
-    , relayDeliveryRates = emptyRMap
-    , postFetchRate = 0
-    , postPublishRate = 0
-    }

src/Hash2Pub/PostService/API.hs

@@ -1,37 +0,0 @@
-{-# LANGUAGE DataKinds             #-}
-{-# LANGUAGE FlexibleInstances     #-}
-{-# LANGUAGE InstanceSigs          #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE OverloadedStrings     #-}
-{-# LANGUAGE RankNTypes            #-}
-{-# LANGUAGE TypeOperators         #-}
-module Hash2Pub.PostService.API where
-
-import           Data.Text.Lazy (Text)
-
-import           Servant
-
-type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Text :> ReqBody '[PlainText] Text :> PutCreated '[PlainText] NoContent
-                 -- delivery endpoint at responsible relay for delivering posts of $tag for distribution
-                 :<|> "relay" :> "subscribers" :> Capture "senderID" Integer :> ReqBody '[PlainText] Text :> PostNoContent '[PlainText] Text
-                 -- endpoint for delivering the subscriptions and outstanding queue
-                 :<|> "post" :> Capture "postid" Text :> Get '[PlainText] Text
-                 -- fetch endpoint for posts, full post ID is http://$domain/post/$postid
-                 :<|> "posts" :> ReqBody '[PlainText] Text :> Post '[PlainText] Text
-                 -- endpoint for fetching multiple posts at once
-                 :<|> "posts" :> "inbox" :> ReqBody '[PlainText] Text :> PutCreated '[PlainText] NoContent
-                 -- delivery endpoint of newly published posts of the relay's instance
-                 :<|> "tags" :> Capture "hashtag" Text :> ReqBody '[PlainText] Text :> PostCreated '[PlainText] Text
-                 -- delivery endpoint for posts of $tag at subscribing instance
-                 :<|> "tags" :> Capture "hashtag" Text :> "subscribe" :> Header "Origin" Text :> Get '[PlainText] Integer
-                 -- endpoint for subscribing the instance specified in
-                 -- the Origin header to $hashtag.
-                 -- Returns subscription lease time in seconds.
-                 :<|> "tags" :> Capture "hashtag" Text :> "unsubscribe" :> Header "Origin" Text :> Get '[PlainText] Text
-                 -- endpoint for unsubscribing the instance specified in
-                 -- the Origin header to $hashtag
-
--- | needed for guiding type inference
-exposedPostServiceAPI :: Proxy PostServiceAPI
-exposedPostServiceAPI = Proxy
-

src/Hash2Pub/ProtocolTypes.hs

@@ -1,5 +1,7 @@
 module Hash2Pub.ProtocolTypes where
 
+import qualified Data.Map                as Map
+import           Data.Maybe              (mapMaybe)
 import qualified Data.Set                as Set
 import           Data.Time.Clock.POSIX   (POSIXTime)
 

src/Hash2Pub/RingMap.hs

@@ -23,30 +23,7 @@ instance (Bounded k, Ord k, Eq a) => Eq (RingMap k a) where
     a == b = getRingMap a == getRingMap b
 
 instance (Bounded k, Ord k, Show k, Show a) => Show (RingMap k a) where
-    show rmap = shows ("RingMap " :: String) (show $ getRingMap rmap)
-
-
-instance (Bounded k, Ord k) => Functor (RingMap k) where
-    -- | map a function over all payload values of a 'RingMap'
-    fmap f = RingMap . Map.map traversingF . getRingMap
-      where
-          traversingF (KeyEntry a) = KeyEntry (f a)
-          traversingF (ProxyEntry pointer (Just entry)) = ProxyEntry pointer (Just $ traversingF entry)
-          traversingF (ProxyEntry pointer Nothing) = ProxyEntry pointer Nothing
-
-
-instance (Bounded k, Ord k) => Foldable (RingMap k) where
-    foldr f initVal = Map.foldr traversingFR initVal . getRingMap
-      where
-          traversingFR (KeyEntry a) acc = f a acc
-          traversingFR (ProxyEntry _ Nothing) acc = acc
-          traversingFR (ProxyEntry _ (Just entry)) acc = traversingFR entry acc
-    foldl f initVal = Map.foldl traversingFL initVal . getRingMap
-      where
-          traversingFL acc (KeyEntry a) = f acc a
-          traversingFL acc (ProxyEntry _ Nothing) = acc
-          traversingFL acc (ProxyEntry _ (Just entry)) = traversingFL acc entry
-
+    show rmap = shows "RingMap " (show $ getRingMap rmap)
 
 -- | entry of a 'RingMap' that holds a value and can also
 -- wrap around the lookup direction at the edges of the name space.
@@ -156,7 +133,7 @@ rMapLookupPred :: (Bounded k, Ord k, Num k)
 rMapLookupPred = lookupWrapper Map.lookupLT Map.lookupLE Backwards
 
 addRMapEntryWith :: (Bounded k, Ord k)
-                 => (RingEntry k a -> RingEntry k a -> RingEntry k a) -- ^ f new_value mold_value
+                 => (RingEntry k a -> RingEntry k a -> RingEntry k a)
                  -> k   -- ^ key
                  -> a   -- ^ value
                  -> RingMap k a
@@ -230,7 +207,7 @@ takeEntriesUntil_ :: (Integral i, Bounded k, Ord k)
                  -> Maybe i     -- possible number limit
                  -> [a]
                  -> [a]
-takeEntriesUntil_ _rmap' _getterFunc' _havingReached _previousEntry (Just remaining) takeAcc
+takeEntriesUntil_ rmap' getterFunc' havingReached previousEntry (Just remaining) takeAcc
   -- length limit reached
   | remaining <= 0 = takeAcc
 takeEntriesUntil_ rmap' getterFunc' havingReached previousEntry numLimit takeAcc =