assistant: XMPP git pull and push requests are cached and sent when presence of a new client is detected.

Noticed that, At startup or network reconnect, git push messages were sent,
often before presence info has been gathered, so were not sent to any
buddies.

To fix this, keep track of which buddies have seen such messages,
and when new presence is received from a buddy that has not yet seen it,
resend.

This is done only for push initiation messages, so very little data needs
to be stored.

5 files changed, 106 insertions, 20 deletions

diff --git a/Assistant/NetMessager.hs b/Assistant/NetMessager.hs
index 7098957b3..97d17af6e 100644
--- a/Assistant/NetMessager.hs
+++ b/Assistant/NetMessager.hs
@@ -15,6 +15,7 @@ import Control.Concurrent.STM
 import Control.Concurrent.MSampleVar
 import Control.Exception as E
 import qualified Data.Set as S
+import qualified Data.Map as M
 
 sendNetMessage :: NetMessage -> Assistant ()
 sendNetMessage m = 
@@ -30,6 +31,42 @@ notifyNetMessagerRestart =
 waitNetMessagerRestart :: Assistant ()
 waitNetMessagerRestart = readSV <<~ (netMessagerRestart . netMessager)
 
+{- Store an important NetMessage for a client, and if the same message was
+ - already sent, remove it from sentImportantNetMessages. -}
+storeImportantNetMessage :: NetMessage -> ClientID -> (ClientID -> Bool) -> Assistant ()
+storeImportantNetMessage m client matchingclient = go <<~ netMessager
+  where
+	go nm = atomically $ do
+		q <- takeTMVar $ importantNetMessages nm
+		sent <- takeTMVar $ sentImportantNetMessages nm
+		putTMVar (importantNetMessages nm) $
+			M.alter (Just . maybe (S.singleton m) (S.insert m)) client q
+		putTMVar (sentImportantNetMessages nm) $
+			M.mapWithKey removematching sent
+	removematching someclient s
+		| matchingclient someclient = S.delete m s
+		| otherwise = s
+
+{- Indicates that an important NetMessage has been sent to a client. -}
+sentImportantNetMessage :: NetMessage -> ClientID -> Assistant ()
+sentImportantNetMessage m client = go <<~ (sentImportantNetMessages . netMessager)
+  where
+	go v = atomically $ do
+		sent <- takeTMVar v
+		putTMVar v $
+			M.alter (Just . maybe (S.singleton m) (S.insert m)) client sent
+
+{- Checks for important NetMessages that have been stored for a client, and
+ - sent to a client. Typically the same client for both, although 
+ - a modified or more specific client may need to be used. -}
+checkImportantNetMessages :: (ClientID, ClientID) -> Assistant (S.Set NetMessage, S.Set NetMessage)
+checkImportantNetMessages (storedclient, sentclient) = go <<~ netMessager
+  where
+	go nm = atomically $ do
+		stored <- M.lookup storedclient <$> (readTMVar $ importantNetMessages nm)
+		sent <- M.lookup sentclient <$> (readTMVar $ sentImportantNetMessages nm)
+		return (fromMaybe S.empty stored, fromMaybe S.empty sent)
+
 {- Runs an action that runs either the send or receive side of a push.
  -
  - While the push is running, netMessagesPush will get messages put into it
diff --git a/Assistant/Threads/XMPPClient.hs b/Assistant/Threads/XMPPClient.hs
index 68806ca4b..0b639cf60 100644
--- a/Assistant/Threads/XMPPClient.hs
+++ b/Assistant/Threads/XMPPClient.hs
@@ -35,7 +35,7 @@ import Data.Time.Clock
 
 {- Whether to include verbose protocol dump in debug output. -}
 protocolDebug :: Bool
-protocolDebug = True
+protocolDebug = False
 
 xmppClientThread :: UrlRenderer -> NamedThread
 xmppClientThread urlrenderer = namedThread "XMPPClient" $
@@ -97,10 +97,10 @@ xmppClient urlrenderer d creds =
 			inAssistant $ debug ["received:", show l]
 		mapM_ (handle selfjid) l
 
-	handle _ (PresenceMessage p) = do
-		
+	handle selfjid (PresenceMessage p) = do
 		void $ inAssistant $ 
 			updateBuddyList (updateBuddies p) <<~ buddyList
+		resendImportantMessages selfjid p
 	handle _ (GotNetMessage QueryPresence) = putStanza gitAnnexSignature
 	handle _ (GotNetMessage (NotifyPush us)) = void $ inAssistant $ pull us
 	handle selfjid (GotNetMessage (PairingNotification stage c u)) =
@@ -114,6 +114,16 @@ xmppClient urlrenderer d creds =
 	handle _ (Unknown _) = noop
 	handle _ (ProtocolError _) = noop
 
+	resendImportantMessages selfjid (Presence { presenceFrom = Just jid }) = do
+		let c = formatJID jid
+		(stored, sent) <- inAssistant $
+			checkImportantNetMessages (formatJID (baseJID jid), c)
+		forM_ (S.toList $ S.difference stored sent) $ \msg -> do
+			inAssistant $ debug ["sending to new client:", show c, show msg]
+			a <- inAssistant $ convertNetMsg (readdressNetMessage msg c) selfjid
+			a
+			inAssistant $ sentImportantNetMessage msg c
+	resendImportantMessages _ _ = noop
 
 data XMPPEvent
 	= GotNetMessage NetMessage
@@ -151,21 +161,27 @@ decodeStanza _ s = [Unknown s]
  - Chat messages must be directed to specific clients, not a base
  - account JID, due to git-annex clients using a negative presence priority.
  - PairingNotification messages are always directed at specific
- - clients, but Pushing messages are sometimes not, and need to be exploded.
+ - clients, but Pushing messages are sometimes not, and need to be exploded
+ - out to specific clients.
+ -
+ - Important messages, not directed at any specific client, 
+ - are cached to be sent later when additional clients connect.
  -}
 relayNetMessage :: JID -> Assistant (XMPP ())
 relayNetMessage selfjid = do
 	msg <- waitNetMessage
 	when protocolDebug $
 		debug ["sending:", show msg]
+	handleImportant msg
 	convert msg
   where
-	convert (NotifyPush us) = return $ putStanza $ pushNotification us
-	convert QueryPresence = return $ putStanza presenceQuery
-	convert (PairingNotification stage c u) = withclient c $ \tojid -> do
-		changeBuddyPairing tojid True
-		return $ putStanza $ pairingNotification stage u tojid selfjid
-	convert (Pushing c pushstage) = withclient c $ \tojid -> do
+	handleImportant msg = case parseJID =<< isImportantNetMessage msg of
+		Just tojid
+			| tojid == baseJID tojid ->
+				storeImportantNetMessage msg (formatJID tojid) $
+					\c -> (baseJID <$> parseJID c) == Just tojid
+		_ -> noop
+	convert (Pushing c pushstage) = withOtherClient selfjid c $ \tojid -> do
 		if tojid == baseJID tojid
 			then do
 				clients <- maybe [] (S.toList . buddyAssistants)
@@ -175,12 +191,29 @@ relayNetMessage selfjid = do
 				return $ forM_ (clients) $ \(Client jid) ->
 					putStanza $ pushMessage pushstage jid selfjid
 			else return $ putStanza $ pushMessage pushstage tojid selfjid
+	convert msg = convertNetMsg msg selfjid
 
-	withclient c a = case parseJID c of
-		Nothing -> return noop
-		Just tojid
-			| tojid == selfjid -> return noop
-			| otherwise -> a tojid
+{- Converts a NetMessage to an XMPP action. -}
+convertNetMsg :: NetMessage -> JID -> Assistant (XMPP ())
+convertNetMsg msg selfjid = convert msg
+  where
+	convert (NotifyPush us) = return $ putStanza $ pushNotification us
+	convert QueryPresence = return $ putStanza presenceQuery
+	convert (PairingNotification stage c u) = withOtherClient selfjid c $ \tojid -> do
+		changeBuddyPairing tojid True
+		return $ putStanza $ pairingNotification stage u tojid selfjid
+	convert (Pushing c pushstage) = withOtherClient selfjid c $ \tojid ->
+		return $ putStanza $  pushMessage pushstage tojid selfjid
+
+withOtherClient :: JID -> ClientID -> (JID -> Assistant (XMPP ())) -> (Assistant (XMPP ()))
+withOtherClient selfjid c a = case parseJID c of
+	Nothing -> return noop
+	Just tojid
+		| tojid == selfjid -> return noop
+		| otherwise -> a tojid
+
+withClient :: ClientID -> (JID -> XMPP ()) -> XMPP ()
+withClient c a = maybe noop a $ parseJID c
 
 {- Runs a XMPP action in a separate thread, using a session to allow it
  - to access the same XMPP client. -}
diff --git a/Assistant/Types/NetMessager.hs b/Assistant/Types/NetMessager.hs
index c036d624a..55bf896bd 100644
--- a/Assistant/Types/NetMessager.hs
+++ b/Assistant/Types/NetMessager.hs
@@ -15,6 +15,7 @@ import Control.Concurrent.STM
 import Control.Concurrent.MSampleVar
 import Data.ByteString (ByteString)
 import qualified Data.Set as S
+import qualified Data.Map as M
 
 {- Messages that can be sent out of band by a network messager. -}
 data NetMessage 
@@ -47,6 +48,18 @@ data PushStage
 	| ReceivePackDone ExitCode
 	deriving (Show, Eq, Ord)
 
+{- NetMessages that are important (and small), and should be stored to be
+ - resent when new clients are seen. -}
+isImportantNetMessage :: NetMessage -> Maybe ClientID
+isImportantNetMessage (Pushing c CanPush) = Just c
+isImportantNetMessage (Pushing c PushRequest) = Just c
+isImportantNetMessage _ = Nothing
+
+readdressNetMessage :: NetMessage -> ClientID -> NetMessage
+readdressNetMessage (PairingNotification stage _ uuid) c = PairingNotification stage c uuid
+readdressNetMessage (Pushing _ stage) c = Pushing c stage
+readdressNetMessage m _ = m
+
 {- Things that initiate either side of a push, but do not actually send data. -}
 isPushInitiation :: PushStage -> Bool
 isPushInitiation CanPush = True
@@ -81,6 +94,10 @@ getSide side m = m side
 data NetMessager = NetMessager
 	-- outgoing messages
 	{ netMessages :: TChan (NetMessage)
+	-- important messages for each client
+	, importantNetMessages :: TMVar (M.Map ClientID (S.Set NetMessage))
+	-- important messages that are believed to have been sent to a client
+	, sentImportantNetMessages :: TMVar (M.Map ClientID (S.Set NetMessage))
 	-- write to this to restart the net messager
 	, netMessagerRestart :: MSampleVar ()
 	-- only one side of a push can be running at a time
@@ -94,8 +111,9 @@ data NetMessager = NetMessager
 newNetMessager :: IO NetMessager
 newNetMessager = NetMessager
 	<$> atomically newTChan
+	<*> atomically (newTMVar M.empty)
+	<*> atomically (newTMVar M.empty)
 	<*> newEmptySV
 	<*> mkSideMap (newTMVar Nothing)
 	<*> mkSideMap newTChan
 	<*> mkSideMap (newTMVar S.empty)
-  where
diff --git a/debian/changelog b/debian/changelog
index 0cdec1fa1..26d4fc76b 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -27,6 +27,8 @@ git-annex (4.20130228) UNRELEASED; urgency=low
     may do undesired things with it.
   * assistant: Get back in sync with XMPP remotes after network reconnection,
     and on startup.
+  * assistant: XMPP git pull and push requests are cached and sent when
+    presence of a new client is detected.
 
  -- Joey Hess <joeyh@debian.org>  Wed, 27 Feb 2013 23:20:40 -0400
 
diff --git a/doc/design/assistant/xmpp.mdwn b/doc/design/assistant/xmpp.mdwn
index 0436fb787..fed79527e 100644
--- a/doc/design/assistant/xmpp.mdwn
+++ b/doc/design/assistant/xmpp.mdwn
@@ -14,10 +14,6 @@ who share a repository, that is stored in the [[cloud]].
   push notification use of XMPP, since unknown UUIDs will just be ignored.
   But XMPP pairing and the pushes over XMPP assume that anyone you're
   paired with is intending to sync to your repository.
-* At startup or network reconnect, git push messages are sent, often before
-  presence info has been gathered, so are not sent to any buddies.
-  Need to cache the last sent CanPush and PushRequest NetMessages for each
-  buddy, and when a new client is seen, resend.
 
 ## design goals