assistant: Get back in sync with XMPP remotes after network reconnection, and on startup.

Make manualPull send push requests over XMPP.

When reconnecting with remotes, those that are XMPP remotes cannot
immediately be pulled from and scanned, so instead maintain a set of
(probably) desynced remotes, and put XMPP remotes on it. (This set could be
used in other ways later, if we can detect we're out of sync with other
types of remotes.)

The merger handles detecting when a XMPP push is received from a desynced
remote, and triggers a scan then, if they have in fact diverged.

This has one known bug: A single XMPP remote can have multiple clients
behind it. When this happens, only the UUID of one client is recorded
as the UUID of the XMPP remote. Pushes from the other XMPP clients will not
trigger a scan. If the client whose UUID is expected responds to the push
request, it'll work, but when that client is offline, we're SOL.

4 files changed, 58 insertions, 31 deletions

diff --git a/Assistant/Sync.hs b/Assistant/Sync.hs
index 8546aa318..f8dcd6748 100644
--- a/Assistant/Sync.hs
+++ b/Assistant/Sync.hs
@@ -18,15 +18,16 @@ import qualified Command.Sync
 import Utility.Parallel
 import qualified Git
 import qualified Git.Branch
-import qualified Git.Ref
 import qualified Git.Command
 import qualified Remote
 import qualified Types.Remote as Remote
 import qualified Annex.Branch
 import Annex.UUID
+import Annex.TaggedPush
 
 import Data.Time.Clock
 import qualified Data.Map as M
+import qualified Data.Set as S
 import Control.Concurrent
 
 {- Syncs with remotes that may have been disconnected for a while.
@@ -36,17 +37,25 @@ import Control.Concurrent
  - An expensive full scan is queued when the git-annex branches of some of
  - the remotes have diverged from the local git-annex branch. Otherwise,
  - it's sufficient to requeue failed transfers.
+ -
+ - XMPP remotes are also signaled that we can push to them, and we request
+ - they push to us. Since XMPP pushes run ansynchronously, any scan of the
+ - XMPP remotes has to be deferred until they're done pushing to us, so
+ - all XMPP remotes are marked as possibly desynced.
  -}
 reconnectRemotes :: Bool -> [Remote] -> Assistant ()
 reconnectRemotes _ [] = noop
 reconnectRemotes notifypushes rs = void $ do
-	alertWhile (syncAlert rs) $ do
+	modifyDaemonStatus_ $ \s -> s
+		{ desynced = S.union (S.fromList $ map Remote.uuid xmppremotes) (desynced s) }
+	alertWhile (syncAlert normalremotes) $ do
 		(ok, diverged) <- sync
 			=<< liftAnnex (inRepo Git.Branch.current)
 		addScanRemotes diverged rs
 		return ok
   where
 	gitremotes = filter (notspecialremote . Remote.repo) rs
+	(xmppremotes, normalremotes) = partition isXMPPRemote gitremotes
 	notspecialremote r
 		| Git.repoIsUrl r = True
 		| Git.repoIsLocal r = True
@@ -128,7 +137,7 @@ pushToRemotes now notifypushes remotes = do
 	fallback branch g u rs = do
 		debug ["fallback pushing to", show rs]
 		(succeeded, failed) <- liftIO $
-			inParallel (\r -> pushFallback u branch r g) rs
+			inParallel (\r -> taggedPush u branch r g) rs
 		updatemap succeeded failed
 		when (notifypushes && (not $ null succeeded)) $
 			sendNetMessage $ NotifyPush $
@@ -137,35 +146,25 @@ pushToRemotes now notifypushes remotes = do
 		
 	push g branch remote = Command.Sync.pushBranch remote branch g
 
-{- This fallback push mode pushes to branches on the remote that have our
- - uuid in them. While ugly, those branches are reserved for pushing by us,
- - and so our pushes will never conflict with other pushes. -}
-pushFallback :: UUID -> Git.Ref -> Remote -> Git.Repo -> IO Bool
-pushFallback u branch remote = Git.Command.runBool
-	[ Param "push"
-	, Param $ Remote.name remote
-	, Param $ refspec Annex.Branch.name
-	, Param $ refspec branch
-	]
-  where
-	{- Push to refs/synced/uuid/branch; this
-	 - avoids cluttering up the branch display. -}
-	refspec b = concat
-		[ s
-		, ":"
-		, "refs/synced/" ++ fromUUID u ++ "/" ++ s
-		]
-	  where s = show $ Git.Ref.base b
-
-{- Manually pull from remotes and merge their branches. -}
+{- Manually pull from remotes and merge their branches. Returns the results
+ - of all the pulls, and whether the git-annex branches of the remotes and
+ - local had divierged before the pull.
+ -
+ - After pulling from the normal git remotes, requests pushes from any XMPP
+ - remotes. However, those pushes will run asynchronously, so their
+ - results are not included in the return data.
+ -}
 manualPull :: Maybe Git.Ref -> [Remote] -> Assistant ([Bool], Bool)
 manualPull currentbranch remotes = do
 	g <- liftAnnex gitRepo
-	results <- liftIO $ forM remotes $ \r ->
+	let (xmppremotes, normalremotes) = partition isXMPPRemote remotes
+	results <- liftIO $ forM normalremotes $ \r ->
 		Git.Command.runBool [Param "fetch", Param $ Remote.name r] g
 	haddiverged <- liftAnnex Annex.Branch.forceUpdate
-	forM_ remotes $ \r ->
+	forM_ normalremotes $ \r ->
 		liftAnnex $ Command.Sync.mergeRemote r currentbranch
+	forM_ xmppremotes $ \r ->
+		sendNetMessage $ Pushing (getXMPPClientID r) PushRequest
 	return (results, haddiverged)
 
 {- Start syncing a newly added remote, using a background thread. -}
diff --git a/Assistant/Threads/Merger.hs b/Assistant/Threads/Merger.hs
index 1488a2f0d..d88cf00bd 100644
--- a/Assistant/Threads/Merger.hs
+++ b/Assistant/Threads/Merger.hs
@@ -10,12 +10,18 @@ module Assistant.Threads.Merger where
 import Assistant.Common
 import Assistant.TransferQueue
 import Assistant.BranchChange
+import Assistant.DaemonStatus
+import Assistant.ScanRemotes
 import Utility.DirWatcher
 import Utility.Types.DirWatcher
 import qualified Annex.Branch
 import qualified Git
 import qualified Git.Branch
 import qualified Command.Sync
+import Annex.TaggedPush
+import Remote (remoteFromUUID)
+
+import qualified Data.Set as S
 
 {- This thread watches for changes to .git/refs/, and handles incoming
  - pushes. -}
@@ -64,13 +70,16 @@ onAdd file
 	| ".lock" `isSuffixOf` file = noop
 	| isAnnexBranch file = do
 		branchChanged
-		whenM (liftAnnex Annex.Branch.forceUpdate) $
-			queueDeferredDownloads "retrying deferred download" Later
+		diverged <- liftAnnex Annex.Branch.forceUpdate
+		when diverged $
+			unlessM handleDesynced $
+				queueDeferredDownloads "retrying deferred download" Later
 	| "/synced/" `isInfixOf` file = do
 		mergecurrent =<< liftAnnex (inRepo Git.Branch.current)
 	| otherwise = noop
   where
 	changedbranch = fileToBranch file
+
 	mergecurrent (Just current)
 		| equivBranches changedbranch current = do
 			debug
@@ -80,6 +89,22 @@ onAdd file
 			void $ liftAnnex  $ Command.Sync.mergeFrom changedbranch
 	mergecurrent _ = noop
 
+	handleDesynced = case branchTaggedBy changedbranch of
+		Nothing -> return False
+		Just u -> do
+			s <- desynced <$> getDaemonStatus				
+			if S.member u s
+				then do
+					modifyDaemonStatus_ $ \st -> st
+						{ desynced = S.delete u s }
+					mr <- liftAnnex $ remoteFromUUID u
+					case mr of
+						Just r -> do
+							addScanRemotes True [r]
+							return True
+						Nothing -> return False
+				else return False
+
 equivBranches :: Git.Ref -> Git.Ref -> Bool
 equivBranches x y = base x == base y
   where
diff --git a/Assistant/Types/DaemonStatus.hs b/Assistant/Types/DaemonStatus.hs
index b60d49edf..7da85daa0 100644
--- a/Assistant/Types/DaemonStatus.hs
+++ b/Assistant/Types/DaemonStatus.hs
@@ -20,6 +20,7 @@ import Control.Concurrent.STM
 import Control.Concurrent.Async
 import Data.Time.Clock.POSIX
 import qualified Data.Map as M
+import qualified Data.Set as S
 
 data DaemonStatus = DaemonStatus
 	-- All the named threads that comprise the daemon,
@@ -44,6 +45,8 @@ data DaemonStatus = DaemonStatus
 	, syncGitRemotes :: [Remote]
 	-- Ordered list of remotes to sync data with
 	, syncDataRemotes :: [Remote]
+	-- List of uuids of remotes that we may have gotten out of sync with.
+	, desynced :: S.Set UUID
 	-- Pairing request that is in progress.
 	, pairingInProgress :: Maybe PairingInProgress
 	-- Broadcasts notifications about all changes to the DaemonStatus
@@ -74,6 +77,7 @@ newDaemonStatus = DaemonStatus
 	<*> pure []
 	<*> pure []
 	<*> pure []
+	<*> pure S.empty
 	<*> pure Nothing
 	<*> newNotificationBroadcaster
 	<*> newNotificationBroadcaster
diff --git a/Assistant/XMPP/Git.hs b/Assistant/XMPP/Git.hs
index 8dba309a8..b5c8e382c 100644
--- a/Assistant/XMPP/Git.hs
+++ b/Assistant/XMPP/Git.hs
@@ -19,6 +19,7 @@ import Assistant.Sync
 import qualified Command.Sync
 import qualified Annex.Branch
 import Annex.UUID
+import Annex.TaggedPush
 import Config
 import Git
 import qualified Git.Branch
@@ -251,7 +252,6 @@ handlePushInitiation :: NetMessage -> Assistant ()
 handlePushInitiation (Pushing cid CanPush) =
 	whenXMPPRemote cid $ 
 		sendNetMessage $ Pushing cid PushRequest
-
 handlePushInitiation (Pushing cid PushRequest) =
 	go =<< liftAnnex (inRepo Git.Branch.current)
   where
@@ -264,8 +264,7 @@ handlePushInitiation (Pushing cid PushRequest) =
 			<*> getUUID
 		liftIO $ Command.Sync.updateBranch (Command.Sync.syncBranch branch) g
 		debug ["pushing to", show rs]
-		forM_ rs $ \r -> xmppPush cid $ pushFallback u branch r
-
+		forM_ rs $ \r -> xmppPush cid $ taggedPush u branch r
 handlePushInitiation (Pushing cid StartingPush) =
 	whenXMPPRemote cid $
 		void $ xmppReceivePack cid