refactor ref change watching

Added to change notification to P2P protocol.

Switched to a TBChan so that a single long-running thread can be
started, and serve perhaps intermittent requests for change
notifications, without buffering all changes in memory.

The P2P runner currently starts up a new thread each times it waits
for a change, but that should allow later reusing a thread. Although
each connection from a peer will still need a new watcher thread to run.

The dependency on stm-chans is more or less free; some stuff in yesod
uses it, so it was already indirectly pulled in when building with the
webapp.

This commit was sponsored by Francois Marier on Patreon.

2 files changed, 22 insertions, 0 deletions

diff --git a/P2P/Annex.hs b/P2P/Annex.hs
index d24e65b0f..e9b59652c 100644
--- a/P2P/Annex.hs
+++ b/P2P/Annex.hs
@@ -16,6 +16,7 @@ module P2P.Annex
 import Annex.Common
 import Annex.Content
 import Annex.Transfer
+import Annex.ChangedRefs
 import P2P.Protocol
 import P2P.IO
 import Logs.Location
@@ -114,6 +115,14 @@ runLocal runmode runner a = case a of
 				protoaction False
 				next
 			Right _ -> runner next
+	WaitRefChange next -> do
+		v <- tryNonAsync $ bracket
+			watchChangedRefs
+			(liftIO . stopWatchingChangedRefs)
+			(liftIO . waitChangedRefs)
+		case v of
+			Left e -> return (Left (show e))
+			Right changedrefs -> runner (next changedrefs)
   where
 	transfer mk k af ta = case runmode of
 		-- Update transfer logs when serving.
diff --git a/P2P/Protocol.hs b/P2P/Protocol.hs
index 03c7c70cf..d8be3ff42 100644
--- a/P2P/Protocol.hs
+++ b/P2P/Protocol.hs
@@ -19,6 +19,7 @@ import Utility.Applicative
 import Utility.PartialPrelude
 import Utility.Metered
 import Git.FilePath
+import Annex.ChangedRefs (ChangedRefs)
 
 import Control.Monad
 import Control.Monad.Free
@@ -50,6 +51,8 @@ data Message
 	| AUTH_FAILURE
 	| CONNECT Service
 	| CONNECTDONE ExitCode
+	| NOTIFYCHANGE
+	| CHANGED ChangedRefs
 	| CHECKPRESENT Key
 	| LOCKCONTENT Key
 	| UNLOCKCONTENT
@@ -70,6 +73,8 @@ instance Proto.Sendable Message where
 	formatMessage AUTH_FAILURE = ["AUTH-FAILURE"]
 	formatMessage (CONNECT service) = ["CONNECT", Proto.serialize service]
 	formatMessage (CONNECTDONE exitcode) = ["CONNECTDONE", Proto.serialize exitcode]
+	formatMessage NOTIFYCHANGE = ["NOTIFYCHANGE"]
+	formatMessage (CHANGED refs) = ["CHANGED", Proto.serialize refs]
 	formatMessage (CHECKPRESENT key) = ["CHECKPRESENT", Proto.serialize key]
 	formatMessage (LOCKCONTENT key) = ["LOCKCONTENT", Proto.serialize key]
 	formatMessage UNLOCKCONTENT = ["UNLOCKCONTENT"]
@@ -89,6 +94,8 @@ instance Proto.Receivable Message where
 	parseCommand "AUTH-FAILURE" = Proto.parse0 AUTH_FAILURE
 	parseCommand "CONNECT" = Proto.parse1 CONNECT
 	parseCommand "CONNECTDONE" = Proto.parse1 CONNECTDONE
+	parseCommand "NOTIFYCHANGE" = Proto.parse0 NOTIFYCHANGE
+	parseCommand "CHANGED" = Proto.parse1 CHANGED
 	parseCommand "CHECKPRESENT" = Proto.parse1 CHECKPRESENT
 	parseCommand "LOCKCONTENT" = Proto.parse1 LOCKCONTENT
 	parseCommand "UNLOCKCONTENT" = Proto.parse0 UNLOCKCONTENT
@@ -227,6 +234,8 @@ data LocalF c
 	-- from being deleted, while running the provided protocol
 	-- action. If unable to lock the content, runs the protocol action
 	-- with False.
+	| WaitRefChange (ChangedRefs -> c)
+	-- ^ Waits for one or more git refs to change and returns them.
 	deriving (Functor)
 
 type Local = Free LocalF
@@ -379,6 +388,10 @@ serveAuthed myuuid = void $ serverLoop handler
 	handler (CONNECT service) = do
 		net $ relayService service
 		return ServerContinue
+	handler NOTIFYCHANGE = do
+		refs <- local waitRefChange
+		net $ sendMessage (CHANGED refs)
+		return ServerContinue
 	handler _ = return ServerUnexpected
 
 sendContent :: Key -> AssociatedFile -> Offset -> MeterUpdate -> Proto Bool