git-annex 10.20240701 → 10.20240731
raw patch · 71 files changed
+4131/−680 lines, 71 filesdep +clockdep +servantdep +servant-clientdep ~stm
Dependencies added: clock, servant, servant-client, servant-client-core, servant-server
Dependency ranges changed: stm
Files
- Annex.hs +5/−4
- Annex/Branch.hs +3/−3
- Annex/BranchState.hs +16/−6
- Annex/Cluster.hs +27/−17
- Annex/Content.hs +176/−22
- Annex/Journal.hs +11/−7
- Annex/Locations.hs +16/−1
- Annex/NumCopies.hs +16/−3
- Annex/Path.hs +5/−1
- Annex/Proxy.hs +142/−41
- Annex/SafeDropProof.hs +34/−0
- Annex/Verify.hs +50/−4
- Annex/WorkerPool.hs +2/−4
- Assistant/Threads/Committer.hs +18/−3
- Assistant/Threads/Watcher.hs +2/−5
- BuildFlags.hs +5/−0
- CHANGELOG +26/−0
- CmdLine/Action.hs +2/−3
- CmdLine/GitAnnex.hs +7/−1
- CmdLine/GitAnnexShell.hs +2/−66
- CmdLine/GitRemoteAnnex.hs +1/−1
- Command/Drop.hs +1/−1
- Command/ExtendCluster.hs +1/−1
- Command/Fsck.hs +1/−1
- Command/Move.hs +8/−5
- Command/P2PHttp.hs +174/−0
- Command/P2PStdIO.hs +17/−16
- Command/TestRemote.hs +3/−3
- Git/Credential.hs +5/−3
- P2P/Annex.hs +43/−40
- P2P/Http.hs +184/−0
- P2P/Http/Client.hs +535/−0
- P2P/Http/Server.hs +478/−0
- P2P/Http/State.hs +657/−0
- P2P/Http/Types.hs +402/−0
- P2P/Http/Url.hs +80/−0
- P2P/IO.hs +54/−29
- P2P/Protocol.hs +184/−38
- P2P/Proxy.hs +217/−93
- Remote/Adb.hs +1/−1
- Remote/BitTorrent.hs +2/−2
- Remote/Bup.hs +1/−1
- Remote/Ddar.hs +1/−1
- Remote/Directory.hs +1/−1
- Remote/External.hs +1/−1
- Remote/GCrypt.hs +5/−5
- Remote/Git.hs +158/−45
- Remote/GitLFS.hs +1/−1
- Remote/Glacier.hs +1/−1
- Remote/Helper/Chunked.hs +3/−3
- Remote/Helper/Hooks.hs +2/−1
- Remote/Helper/P2P.hs +32/−21
- Remote/Helper/ReadOnly.hs +2/−2
- Remote/Helper/Special.hs +6/−6
- Remote/Helper/Ssh.hs +13/−9
- Remote/Hook.hs +1/−1
- Remote/Rsync.hs +1/−1
- Remote/S3.hs +2/−2
- Remote/Tahoe.hs +3/−3
- Remote/Web.hs +2/−2
- Remote/WebDAV.hs +1/−1
- Types/GitConfig.hs +101/−74
- Types/NumCopies.hs +80/−25
- Types/Remote.hs +10/−3
- Types/StoreRetrieve.hs +4/−2
- Types/UUID.hs +5/−0
- Upgrade/V1.hs +14/−10
- Utility/MonotonicClock.hs +37/−0
- git-annex.cabal +27/−3
- stack-lts-18.13.yaml +0/−28
- stack.yaml +3/−2
Annex.hs view
@@ -115,7 +115,8 @@ -- Values that can be read, but not modified by an Annex action. data AnnexRead = AnnexRead- { activekeys :: TVar (M.Map Key ThreadId)+ { branchstate :: MVar BranchState+ , activekeys :: TVar (M.Map Key ThreadId) , activeremotes :: MVar (M.Map (Types.Remote.RemoteA Annex) Integer) , keysdbhandle :: Keys.DbHandle , sshstalecleaned :: TMVar Bool@@ -137,6 +138,7 @@ newAnnexRead :: GitConfig -> IO AnnexRead newAnnexRead c = do+ bs <- newMVar startBranchState emptyactivekeys <- newTVarIO M.empty emptyactiveremotes <- newMVar M.empty kh <- Keys.newDbHandle@@ -146,7 +148,8 @@ cm <- newTMVarIO M.empty cc <- newTMVarIO (CredentialCache M.empty) return $ AnnexRead- { activekeys = emptyactivekeys+ { branchstate = bs+ , activekeys = emptyactivekeys , activeremotes = emptyactiveremotes , keysdbhandle = kh , sshstalecleaned = sc@@ -180,7 +183,6 @@ , output :: MessageState , concurrency :: ConcurrencySetting , daemon :: Bool- , branchstate :: BranchState , repoqueue :: Maybe (Git.Queue.Queue Annex) , catfilehandles :: CatFileHandles , hashobjecthandle :: Maybe (ResourcePool HashObjectHandle)@@ -235,7 +237,6 @@ , output = o , concurrency = ConcurrencyCmdLine NonConcurrent , daemon = False- , branchstate = startBranchState , repoqueue = Nothing , catfilehandles = catFileHandlesNonConcurrent , hashobjecthandle = Nothing
Annex/Branch.hs view
@@ -262,7 +262,7 @@ else commitIndex jl branchref merge_desc commitrefs ) addMergedRefs tomerge- invalidateCache+ invalidateCacheAll stagejournalwhen dirty jl a | dirty = stageJournal jl a@@ -487,7 +487,7 @@ -- evaluating a Journalable Builder twice, which is not very -- efficient. Instead, assume that it's not common to need to read -- a log file immediately after writing it.- invalidateCache+ invalidateCache f {- Appends content to the journal file. -} append :: Journalable content => JournalLocked -> RawFilePath -> AppendableJournalFile -> content -> Annex ()@@ -495,7 +495,7 @@ journalChanged appendJournalFile jl appendable toappend fastDebug "Annex.Branch" ("append " ++ fromRawFilePath f)- invalidateCache+ invalidateCache f {- Commit message used when making a commit of whatever data has changed - to the git-annex branch. -}
Annex/BranchState.hs view
@@ -2,7 +2,7 @@ - - Runtime state about the git-annex branch, and a small cache. -- - Copyright 2011-2022 Joey Hess <id@joeyh.name>+ - Copyright 2011-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -16,14 +16,18 @@ import Logs import qualified Git +import Control.Concurrent import qualified Data.ByteString.Lazy as L getState :: Annex BranchState-getState = Annex.getState Annex.branchstate+getState = do+ v <- Annex.getRead Annex.branchstate+ liftIO $ readMVar v changeState :: (BranchState -> BranchState) -> Annex ()-changeState changer = Annex.changeState $ \s -> - s { Annex.branchstate = changer (Annex.branchstate s) }+changeState changer = do+ v <- Annex.getRead Annex.branchstate+ liftIO $ modifyMVar_ v $ return . changer {- Runs an action to check that the index file exists, if it's not been - checked before in this run of git-annex. -}@@ -130,5 +134,11 @@ | f == file && not (needInteractiveAccess state) = Just c | otherwise = go rest -invalidateCache :: Annex ()-invalidateCache = changeState $ \s -> s { cachedFileContents = [] }+invalidateCache :: RawFilePath -> Annex ()+invalidateCache f = changeState $ \s -> s+ { cachedFileContents = filter (\(f', _) -> f' /= f) + (cachedFileContents s)+ }++invalidateCacheAll :: Annex ()+invalidateCacheAll = changeState $ \s -> s { cachedFileContents = [] }
Annex/Cluster.hs view
@@ -18,6 +18,7 @@ import P2P.IO import Annex.Proxy import Annex.UUID+import Annex.BranchState import Logs.Location import Logs.PreferredContent import Types.Command@@ -38,22 +39,17 @@ -> (forall a. Annex () -> ((a -> CommandPerform) -> Annex (Either ProtoFailure a) -> CommandPerform)) -> CommandPerform proxyCluster clusteruuid proxydone servermode clientside protoerrhandler = do+ enableInteractiveBranchAccess getClientProtocolVersion (fromClusterUUID clusteruuid) clientside withclientversion (protoerrhandler noop) where- proxymethods = ProxyMethods- { removedContent = \u k -> logChange k u InfoMissing- , addedContent = \u k -> logChange k u InfoPresent- }- withclientversion (Just (clientmaxversion, othermsg)) = do -- The protocol versions supported by the nodes are not -- known at this point, and would be too expensive to -- determine. Instead, pick the newest protocol version -- that we and the client both speak. The proxy code- -- checks protocol versions when operating on multiple- -- nodes, and allows nodes to have different protocol- -- versions.+ -- checks protocol versions of remotes, so nodes can+ -- have different protocol versions. let protocolversion = min maxProtocolVersion clientmaxversion sendClientProtocolVersion clientside othermsg protocolversion (getclientbypass protocolversion) (protoerrhandler noop)@@ -64,15 +60,29 @@ (withclientbypass protocolversion) (protoerrhandler noop) withclientbypass protocolversion (bypassuuids, othermsg) = do- (selectnode, closenodes) <- clusterProxySelector clusteruuid- protocolversion bypassuuids- concurrencyconfig <- getConcurrencyConfig- proxy proxydone proxymethods servermode clientside - (fromClusterUUID clusteruuid)- selectnode concurrencyconfig protocolversion- othermsg (protoerrhandler closenodes)+ (selectnode, closenodes) <-+ clusterProxySelector clusteruuid+ protocolversion bypassuuids+ proxystate <- liftIO mkProxyState+ concurrencyconfig <- concurrencyConfigJobs+ let proxyparams = ProxyParams+ { proxyMethods = mkProxyMethods+ , proxyState = proxystate+ , proxyServerMode = servermode+ , proxyClientSide = clientside+ , proxyUUID = fromClusterUUID clusteruuid+ , proxySelector = selectnode+ , proxyConcurrencyConfig = concurrencyconfig+ , proxyClientProtocolVersion = protocolversion+ }+ proxy proxydone proxyparams othermsg+ (protoerrhandler closenodes) -clusterProxySelector :: ClusterUUID -> ProtocolVersion -> Bypass -> Annex (ProxySelector, Annex ())+clusterProxySelector+ :: ClusterUUID+ -> ProtocolVersion+ -> Bypass+ -> Annex (ProxySelector, Annex ()) clusterProxySelector clusteruuid protocolversion (Bypass bypass) = do nodeuuids <- (fromMaybe S.empty . M.lookup clusteruuid . clusterUUIDs) <$> getClusters@@ -107,11 +117,11 @@ -- could be out of date, actually try to remove from every -- node. , proxyREMOVE = const (pure nodes)+ , proxyGETTIMESTAMP = pure nodes -- Content is not locked on the cluster as a whole, -- instead it can be locked on individual nodes that are -- proxied to the client. , proxyLOCKCONTENT = const (pure Nothing)- , proxyUNLOCKCONTENT = pure Nothing } return (proxyselector, closenodes) where
Annex/Content.hs view
@@ -1,6 +1,6 @@ {- git-annex file content managing -- - Copyright 2010-2023 Joey Hess <id@joeyh.name>+ - Copyright 2010-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -88,6 +88,7 @@ import Annex.Perms import Annex.Link import Annex.LockPool+import Annex.LockFile import Annex.UUID import Annex.InodeSentinal import Annex.ReplaceFile@@ -103,6 +104,8 @@ import Utility.InodeCache import Utility.CopyFile import Utility.Metered+import Utility.HumanTime+import Utility.TimeStamp #ifndef mingw32_HOST_OS import Utility.FileMode #endif@@ -110,38 +113,104 @@ import qualified System.FilePath.ByteString as P import System.PosixCompat.Files (isSymbolicLink, linkCount)+import Data.Time.Clock.POSIX {- Prevents the content from being removed while the action is running. - Uses a shared lock. - - If locking fails, or the content is not present, throws an exception - rather than running the action.+ -+ - When a Duration is provided, the content is prevented from being removed+ - for that amount of time, even if the current process is terminated.+ - (This is only done when using a separate lock file from the content+ - file eg in v10 and higher repositories.) -}-lockContentShared :: Key -> (VerifiedCopy -> Annex a) -> Annex a-lockContentShared key a = lockContentUsing lock key notpresent $- ifM (inAnnex key)- ( do- u <- getUUID- withVerifiedCopy LockedCopy u (return True) a- , notpresent- )+lockContentShared :: Key -> Maybe Duration -> (VerifiedCopy -> Annex a) -> Annex a+lockContentShared key mduration a = do+ retention <- case mduration of+ Nothing -> pure Nothing+ Just duration -> do+ rt <- calcRepo (gitAnnexContentRetentionTimestamp key)+ now <- liftIO getPOSIXTime+ pure $ Just+ ( rt+ , now + fromIntegral (durationSeconds duration)+ )+ lockContentUsing (lock retention) key notpresent $+ ifM (inAnnex key)+ ( do+ u <- getUUID+ withVerifiedCopy LockedCopy u (return (Right True)) a+ , notpresent+ ) where notpresent = giveup $ "failed to lock content: not present" #ifndef mingw32_HOST_OS- lock _ (Just lockfile) = - ( posixLocker tryLockShared lockfile- , Just (posixLocker tryLockExclusive lockfile)+ lock retention _ (Just lockfile) = + ( posixLocker tryLockShared lockfile >>= \case+ Just lck -> do+ writeretention retention+ return (Just lck)+ Nothing -> return Nothing+ , Just $ posixLocker tryLockExclusive lockfile >>= \case+ Just lck -> do+ dropretention retention+ return (Just lck)+ Nothing -> return Nothing )- lock contentfile Nothing =+ lock _ contentfile Nothing = ( tryLockShared Nothing contentfile , Nothing ) #else- lock = winLocker lockShared+ lock retention obj lckf = + let (locker, postunlock) = winLocker lockShared obj lckf+ in + ( locker >>= \case+ Just lck -> do+ writeretention retention+ return (Just lck)+ Nothing -> return Nothing+ , Just $ \lckfile -> do+ maybe noop (\pu -> pu lckfile) postunlock+ lockdropretention obj lckf retention+ )++ lockdropretention _ _ Nothing = noop+ lockdropretention obj lckf retention = do+ -- In order to dropretention, have to+ -- take an exclusive lock.+ let (exlocker, expostunlock) =+ winLocker lockExclusive obj lckf+ exlocker >>= \case+ Nothing -> noop+ Just lck -> do+ dropretention retention+ liftIO $ dropLock lck+ case (expostunlock, lckf) of+ (Just pu, Just f) -> pu f+ _ -> noop #endif+ + writeretention Nothing = noop+ writeretention (Just (rt, retentionts)) = + writeContentRetentionTimestamp key rt retentionts+ + -- When this is called, an exclusive lock has been taken, so no other+ -- processes can be writing to the retention time stamp file.+ -- The timestamp in the file may have been written by this+ -- call to lockContentShared or a later call. Only delete the file+ -- in the former case.+ dropretention Nothing = noop+ dropretention (Just (rt, retentionts)) =+ readContentRetentionTimestamp rt >>= \case+ Just ts | ts == retentionts ->+ removeRetentionTimeStamp key rt+ _ -> noop -{- Exclusively locks content, while performing an action that- - might remove it.+{- Exclusively locks content, including checking the retention timestamp, + - while performing an action that might remove it. - - If locking fails, throws an exception rather than running the action. -@@ -155,7 +224,11 @@ a (ContentRemovalLock key) where #ifndef mingw32_HOST_OS- lock _ (Just lockfile) = (posixLocker tryLockExclusive lockfile, Nothing)+ lock _ (Just lockfile) =+ ( checkRetentionTimestamp key+ (posixLocker tryLockExclusive lockfile)+ , Nothing+ ) {- No lock file, so the content file itself is locked. - Since content files are stored with the write bit - disabled, have to fiddle with permissions to open@@ -167,12 +240,32 @@ (tryLockExclusive Nothing contentfile) in (lck, Nothing) #else- lock = winLocker lockExclusive+ lock obj lckf = + let (exlocker, expostunlock) =+ winLocker lockExclusive obj lckf+ in (checkRetentionTimestamp key exlocker, expostunlock) #endif {- Passed the object content file, and maybe a separate lock file to use, - when the content file itself should not be locked. -}-type ContentLocker = RawFilePath -> Maybe LockFile -> (Annex (Maybe LockHandle), Maybe (Annex (Maybe LockHandle)))+type ContentLocker + = RawFilePath+ -> Maybe LockFile + ->+ ( Annex (Maybe LockHandle)+ -- ^ Takes the lock, which may be shared or exclusive.+#ifndef mingw32_HOST_OS+ , Maybe (Annex (Maybe LockHandle))+ -- ^ When the above takes a shared lock, this is used+ -- to take an exclusive lock, after dropping the shared lock,+ -- and prior to deleting the lock file, in order to + -- ensure that no other processes also have a shared lock.+#else+ , Maybe (RawFilePath -> Annex ())+ -- ^ On Windows, this is called after the lock is dropped,+ -- but before the lock file is cleaned up.+#endif+ ) #ifndef mingw32_HOST_OS posixLocker :: (Maybe ModeSetter -> LockFile -> Annex (Maybe LockHandle)) -> LockFile -> Annex (Maybe LockHandle)@@ -264,13 +357,17 @@ maybe noop cleanuplockfile mlockfile liftIO $ dropLock lck #else- unlock _ mlockfile lck = do+ unlock postunlock mlockfile lck = do -- Can't delete a locked file on Windows, -- so close our lock first. If there are other shared- -- locks, they will prevent the file deletion from+ -- locks, they will prevent the lock file deletion from -- happening. liftIO $ dropLock lck- maybe noop cleanuplockfile mlockfile+ case mlockfile of+ Nothing -> noop -- never reached+ Just lockfile -> do+ maybe noop (\pu -> pu lockfile) postunlock+ cleanuplockfile lockfile #endif cleanuplockfile lockfile = void $ tryNonAsync $ do@@ -960,3 +1057,60 @@ ( isUnmodifiedCheap' key ic , return True )++{- Avoids writing a timestamp when the file already contains a later+ - timestamp. The file is written atomically, so when it contained an+ - earlier timestamp, a reader will always see one or the other timestamp.+ -}+writeContentRetentionTimestamp :: Key -> RawFilePath -> POSIXTime -> Annex ()+writeContentRetentionTimestamp key rt t = do+ lckfile <- calcRepo (gitAnnexContentRetentionTimestampLock key)+ modifyContentDirWhenExists lckfile $ bracket (lock lckfile) unlock $ \_ ->+ readContentRetentionTimestamp rt >>= \case+ Just ts | ts >= t -> return ()+ _ -> replaceFile (const noop) (fromRawFilePath rt) $ \tmp ->+ liftIO $ writeFile (fromRawFilePath tmp) $ show t+ where+ lock = takeExclusiveLock+ unlock = liftIO . dropLock++{- Does not need locking because the file is written atomically. -}+readContentRetentionTimestamp :: RawFilePath -> Annex (Maybe POSIXTime)+readContentRetentionTimestamp rt =+ liftIO $ join <$> tryWhenExists + (parsePOSIXTime <$> readFile (fromRawFilePath rt))++{- Checks if the retention timestamp is in the future, if so returns+ - Nothing.+ -+ - If the retention timestamp is in the past, the retention timestamp file+ - is deleted. This cleans up stale retention timestamps.+ -+ - The locker should take a lock that prevents any other processes from+ - writing to the retention timestamp. So the retention timestamp lock+ - is not used here and can also be deleted when deleting the retention+ - timestamp file.+ -}+checkRetentionTimestamp :: Key -> Annex (Maybe LockHandle) -> Annex (Maybe LockHandle)+checkRetentionTimestamp key locker = do+ rt <- calcRepo (gitAnnexContentRetentionTimestamp key)+ readContentRetentionTimestamp rt >>= \case+ Nothing -> locker+ Just ts -> do+ now <- liftIO getPOSIXTime+ if now > ts+ then locker >>= \case+ Nothing -> return Nothing+ Just lock -> do+ removeRetentionTimeStamp key rt+ return (Just lock)+ else return Nothing++{- Remove the retention timestamp and its lock file. Another lock must+ - be held, that prevents anything else writing to the file at the same+ - time. -}+removeRetentionTimeStamp :: Key -> RawFilePath -> Annex ()+removeRetentionTimeStamp key rt = modifyContentDirWhenExists rt $ do+ liftIO $ removeWhenExistsWith R.removeLink rt+ rtl <- calcRepo (gitAnnexContentRetentionTimestampLock key)+ liftIO $ removeWhenExistsWith R.removeLink rtl
Annex/Journal.hs view
@@ -181,11 +181,13 @@ getJournalFileStale :: GetPrivate -> RawFilePath -> Annex JournalledContent getJournalFileStale (GetPrivate getprivate) file = do st <- Annex.getState id+ let repo = Annex.repo st+ bs <- getState liftIO $ if getprivate && privateUUIDsKnown' st then do- x <- getfrom (gitAnnexJournalDir (Annex.branchstate st) (Annex.repo st))- getfrom (gitAnnexPrivateJournalDir (Annex.branchstate st) (Annex.repo st)) >>= \case+ x <- getfrom (gitAnnexJournalDir bs repo)+ getfrom (gitAnnexPrivateJournalDir bs repo) >>= \case Nothing -> return $ case x of Nothing -> NoJournalledContent Just b -> JournalledContent b@@ -195,7 +197,7 @@ -- happens in a merge of two -- git-annex branches. Just x' -> x' <> y- else getfrom (gitAnnexJournalDir (Annex.branchstate st) (Annex.repo st)) >>= return . \case+ else getfrom (gitAnnexJournalDir bs repo) >>= return . \case Nothing -> NoJournalledContent Just b -> JournalledContent b where@@ -223,8 +225,9 @@ - journal is staged as it is run. -} getJournalledFilesStale :: (BranchState -> Git.Repo -> RawFilePath) -> Annex [RawFilePath] getJournalledFilesStale getjournaldir = do- st <- Annex.getState id- let d = getjournaldir (Annex.branchstate st) (Annex.repo st)+ bs <- getState+ repo <- Annex.gitRepo+ let d = getjournaldir bs repo fs <- liftIO $ catchDefaultIO [] $ getDirectoryContents (fromRawFilePath d) return $ filter (`notElem` [".", ".."]) $@@ -233,8 +236,9 @@ {- Directory handle open on a journal directory. -} withJournalHandle :: (BranchState -> Git.Repo -> RawFilePath) -> (DirectoryHandle -> IO a) -> Annex a withJournalHandle getjournaldir a = do- st <- Annex.getState id- let d = getjournaldir (Annex.branchstate st) (Annex.repo st)+ bs <- getState+ repo <- Annex.gitRepo+ let d = getjournaldir bs repo bracket (opendir d) (liftIO . closeDirectory) (liftIO . a) where -- avoid overhead of creating the journal directory when it already
Annex/Locations.hs view
@@ -1,6 +1,6 @@ {- git-annex file locations -- - Copyright 2010-2023 Joey Hess <id@joeyh.name>+ - Copyright 2010-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -20,6 +20,8 @@ gitAnnexLink, gitAnnexLinkCanonical, gitAnnexContentLock,+ gitAnnexContentRetentionTimestamp,+ gitAnnexContentRetentionTimestampLock, gitAnnexContentLockLock, gitAnnexInodeSentinal, gitAnnexInodeSentinalCache,@@ -253,6 +255,19 @@ gitAnnexContentLock key r config = do loc <- gitAnnexLocation key r config return $ loc <> ".lck"++{- File used to indicate a key's content should not be dropped until after+ - a specified time. -}+gitAnnexContentRetentionTimestamp :: Key -> Git.Repo -> GitConfig -> IO RawFilePath+gitAnnexContentRetentionTimestamp key r config = do+ loc <- gitAnnexLocation key r config+ return $ loc <> ".rtm"++{- Lock file for gitAnnexContentRetentionTimestamp -}+gitAnnexContentRetentionTimestampLock :: Key -> Git.Repo -> GitConfig -> IO RawFilePath+gitAnnexContentRetentionTimestampLock key r config = do+ loc <- gitAnnexLocation key r config+ return $ loc <> ".rtl" {- Lock that is held when taking the gitAnnexContentLock to support the v10 - upgrade.
Annex/NumCopies.hs view
@@ -29,6 +29,7 @@ import Annex.Common import qualified Annex+import Annex.SafeDropProof import Types.NumCopies import Logs.NumCopies import Logs.Trust@@ -227,6 +228,10 @@ {- Verifies that enough copies of a key exist among the listed remotes, - to safely drop it, running an action with a proof if so, and - printing an informative message if not.+ -+ - Note that the proof is checked to still be valid at the current time+ - before running the action, but when dropping the key may take some time,+ - the proof's time may need to be checked again. -} verifyEnoughCopiesToDrop :: String -- message to print when there are no known locations@@ -246,17 +251,17 @@ where helper bad missing have [] lockunsupported = liftIO (mkSafeDropProof neednum needmin have removallock) >>= \case- Right proof -> dropaction proof+ Right proof -> checkprooftime proof Left stillhave -> do notEnoughCopies key dropfrom neednum needmin stillhave (skip++missing) bad nolocmsg lockunsupported nodropaction helper bad missing have (c:cs) lockunsupported | isSafeDrop neednum needmin have removallock = liftIO (mkSafeDropProof neednum needmin have removallock) >>= \case- Right proof -> dropaction proof+ Right proof -> checkprooftime proof Left stillhave -> helper bad missing stillhave (c:cs) lockunsupported | otherwise = case c of- UnVerifiedHere -> lockContentShared key contverified+ UnVerifiedHere -> lockContentShared key Nothing contverified UnVerifiedRemote r -- Skip cluster uuids because locking is -- not supported with them, instead will@@ -294,6 +299,14 @@ , MC.Handler (\ (_e :: SomeException) -> fallback) ] Nothing -> fallback+ + checkprooftime proof = + ifM (liftIO $ checkSafeDropProofEndTime (Just proof))+ ( dropaction proof+ , do+ safeDropProofExpired+ nodropaction+ ) data DropException = DropException SomeException deriving (Typeable, Show)
Annex/Path.hs view
@@ -85,7 +85,11 @@ gitAnnexChildProcessParams :: String -> [CommandParam] -> Annex [CommandParam] gitAnnexChildProcessParams subcmd ps = do cps <- gitAnnexGitConfigOverrides- return (Param subcmd : cps ++ ps)+ force <- Annex.getRead Annex.force+ let cps' = if force+ then Param "--force" : cps+ else cps+ return (Param subcmd : cps' ++ ps) gitAnnexGitConfigOverrides :: Annex [CommandParam] gitAnnexGitConfigOverrides = concatMap (\c -> [Param "-c", Param c])
Annex/Proxy.hs view
@@ -8,23 +8,31 @@ module Annex.Proxy where import Annex.Common-import P2P.Proxy-import P2P.Protocol-import P2P.IO+import qualified Annex import qualified Remote import qualified Types.Remote as Remote import qualified Remote.Git+import P2P.Proxy+import P2P.Protocol+import P2P.IO import Remote.Helper.Ssh (openP2PShellConnection', closeP2PShellConnection)-import Annex.Content import Annex.Concurrent import Annex.Tmp+import Annex.Verify+import Logs.Proxy+import Logs.Cluster+import Logs.UUID+import Logs.Location import Utility.Tmp.Dir import Utility.Metered import Control.Concurrent.STM import Control.Concurrent.Async+import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L import qualified System.FilePath.ByteString as P+import qualified Data.Map as M+import qualified Data.Set as S proxyRemoteSide :: ProtocolVersion -> Bypass -> Remote -> Annex RemoteSide proxyRemoteSide clientmaxversion bypass r@@ -53,18 +61,19 @@ ohdl <- liftIO newEmptyTMVarIO iwaitv <- liftIO newEmptyTMVarIO owaitv <- liftIO newEmptyTMVarIO- endv <- liftIO newEmptyTMVarIO+ iclosedv <- liftIO newEmptyTMVarIO+ oclosedv <- liftIO newEmptyTMVarIO worker <- liftIO . async =<< forkState- (proxySpecialRemote protoversion r ihdl ohdl owaitv endv)+ (proxySpecialRemote protoversion r ihdl ohdl owaitv oclosedv) let remoteconn = P2PConnection { connRepo = Nothing , connCheckAuth = const False- , connIhdl = P2PHandleTMVar ihdl iwaitv- , connOhdl = P2PHandleTMVar ohdl owaitv+ , connIhdl = P2PHandleTMVar ihdl (Just iwaitv) iclosedv+ , connOhdl = P2PHandleTMVar ohdl (Just owaitv) oclosedv , connIdent = ConnIdent (Just (Remote.name r)) } let closeremoteconn = do- liftIO $ atomically $ putTMVar endv ()+ liftIO $ atomically $ putTMVar oclosedv () join $ liftIO (wait worker) return $ Just ( remoterunst@@ -81,7 +90,7 @@ -> TMVar () -> TMVar () -> Annex ()-proxySpecialRemote protoversion r ihdl ohdl owaitv endv = go+proxySpecialRemote protoversion r ihdl ohdl owaitv oclosedv = go where go :: Annex () go = liftIO receivemessage >>= \case@@ -96,7 +105,7 @@ liftIO $ sendmessage FAILURE go Just (REMOVE k) -> do- tryNonAsync (Remote.removeKey r k) >>= \case+ tryNonAsync (Remote.removeKey r Nothing k) >>= \case Right () -> liftIO $ sendmessage SUCCESS Left err -> liftIO $ propagateerror err go@@ -114,23 +123,28 @@ liftIO $ sendmessage $ ERROR "NOTIFYCHANGE unsupported for a special remote" go- Just _ -> giveup "protocol error M"+ Just _ -> giveup "protocol error" Nothing -> return () - getnextmessageorend = - liftIO $ atomically $ - (Right <$> takeTMVar ohdl)- `orElse`- (Left <$> readTMVar endv)-- receivemessage = getnextmessageorend >>= \case+ receivemessage = liftIO (atomically recv) >>= \case Right (Right m) -> return (Just m) Right (Left _b) -> giveup "unexpected ByteString received from P2P MVar" Left () -> return Nothing+ where+ recv = + (Right <$> takeTMVar ohdl)+ `orElse`+ (Left <$> readTMVar oclosedv) - receivebytestring = atomically (takeTMVar ohdl) >>= \case- Left b -> return b- Right _m -> giveup "did not receive ByteString from P2P MVar"+ receivebytestring = atomically recv >>= \case+ Right (Left b) -> return (Just b)+ Right (Right _m) -> giveup "did not receive ByteString from P2P MVar"+ Left () -> return Nothing+ where+ recv = + (Right <$> takeTMVar ohdl)+ `orElse`+ (Left <$> readTMVar oclosedv) sendmessage m = atomically $ putTMVar ihdl (Right m) @@ -148,6 +162,8 @@ withTmpDirIn (fromRawFilePath othertmpdir) "proxy" $ \tmpdir -> a (toRawFilePath tmpdir P.</> keyFile k) + -- Verify the content received from the client, to avoid bad content+ -- being stored in the special remote. proxyput af k = do liftIO $ sendmessage $ PUT_FROM (Offset 0) withproxytmpfile k $ \tmpfile -> do@@ -155,33 +171,54 @@ Right () -> liftIO $ sendmessage SUCCESS Left err -> liftIO $ propagateerror err liftIO receivemessage >>= \case- Just (DATA (Len _)) -> do- b <- liftIO receivebytestring- liftIO $ L.writeFile (fromRawFilePath tmpfile) b- -- Signal that the whole bytestring- -- has been received.- liftIO $ atomically $ putTMVar owaitv ()+ Just (DATA (Len len)) -> do+ iv <- startVerifyKeyContentIncrementally Remote.AlwaysVerify k+ h <- liftIO $ openFile (fromRawFilePath tmpfile) WriteMode+ gotall <- liftIO $ receivetofile iv h len+ liftIO $ hClose h+ verified <- if gotall+ then fst <$> finishVerifyKeyContentIncrementally' True iv+ else pure False if protoversion > ProtocolVersion 1 then liftIO receivemessage >>= \case- Just (VALIDITY Valid) ->- store+ Just (VALIDITY Valid)+ | verified -> store+ | otherwise -> liftIO $ sendmessage FAILURE Just (VALIDITY Invalid) ->- return ()- _ -> giveup "protocol error N"+ liftIO $ sendmessage FAILURE+ _ -> giveup "protocol error" else store- _ -> giveup "protocol error O"+ _ -> giveup "protocol error"+ liftIO $ removeWhenExistsWith removeFile (fromRawFilePath tmpfile) + receivetofile iv h n = liftIO receivebytestring >>= \case+ Just b -> do+ liftIO $ atomically $ + putTMVar owaitv ()+ `orElse`+ readTMVar oclosedv+ n' <- storetofile iv h n (L.toChunks b)+ -- Normally all the data is sent in a single+ -- lazy bytestring. However, when the special+ -- remote is a node in a cluster, a PUT is+ -- streamed to it in multiple chunks.+ if n' == 0 + then return True+ else receivetofile iv h n'+ Nothing -> return False++ storetofile _ _ n [] = pure n+ storetofile iv h n (b:bs) = do+ writeVerifyChunk iv h b+ B.hPut h b+ storetofile iv h (n - fromIntegral (B.length b)) bs+ proxyget offset af k = withproxytmpfile k $ \tmpfile -> do -- Don't verify the content from the remote, -- because the client will do its own verification. let vc = Remote.NoVerify tryNonAsync (Remote.retrieveKeyFile r k af (fromRawFilePath tmpfile) nullMeterUpdate vc) >>= \case- Right v ->- ifM (verifyKeyContentPostRetrieval Remote.RetrievalVerifiableKeysSecure vc v k tmpfile)- ( liftIO $ senddata offset tmpfile- , liftIO $ sendmessage $- ERROR "verification of content failed"- )+ Right _ -> liftIO $ senddata offset tmpfile Left err -> liftIO $ propagateerror err senddata (Offset offset) f = do@@ -206,6 +243,70 @@ receivemessage >>= \case Just SUCCESS -> return () Just FAILURE -> return ()- Just _ -> giveup "protocol error P"+ Just _ -> giveup "protocol error" Nothing -> return ()- ++{- Check if this repository can proxy for a specified remote uuid,+ - and if so enable proxying for it. -}+checkCanProxy :: UUID -> UUID -> Annex Bool+checkCanProxy remoteuuid ouruuid = do+ ourproxies <- M.lookup ouruuid <$> getProxies+ checkCanProxy' ourproxies remoteuuid >>= \case+ Right v -> do+ Annex.changeState $ \st -> st { Annex.proxyremote = Just v }+ return True+ Left Nothing -> return False+ Left (Just err) -> giveup err++checkCanProxy' :: Maybe (S.Set Proxy) -> UUID -> Annex (Either (Maybe String) (Either ClusterUUID Remote))+checkCanProxy' Nothing _ = return (Left Nothing)+checkCanProxy' (Just proxies) remoteuuid =+ case filter (\p -> proxyRemoteUUID p == remoteuuid) (S.toList proxies) of+ [] -> notconfigured+ ps -> case mkClusterUUID remoteuuid of+ Just cu -> proxyforcluster cu+ Nothing -> proxyfor ps+ where+ -- This repository may have multiple remotes that access the same+ -- repository. Proxy for the lowest cost one that is configured to+ -- be used as a proxy.+ proxyfor ps = do+ rs <- concat . Remote.byCost <$> Remote.remoteList+ myclusters <- annexClusters <$> Annex.getGitConfig+ let sameuuid r = Remote.uuid r == remoteuuid+ let samename r p = Remote.name r == proxyRemoteName p+ case headMaybe (filter (\r -> sameuuid r && proxyisconfigured rs myclusters r && any (samename r) ps) rs) of+ Nothing -> notconfigured+ Just r -> return (Right (Right r))+ + -- Only proxy for a remote when the git configuration+ -- allows it. This is important to prevent changes to + -- the git-annex branch causing unexpected proxying for remotes.+ proxyisconfigured rs myclusters r+ | remoteAnnexProxy (Remote.gitconfig r) = True+ -- Proxy for remotes that are configured as cluster nodes.+ | any (`M.member` myclusters) (fromMaybe [] $ remoteAnnexClusterNode $ Remote.gitconfig r) = True+ -- Proxy for a remote when it is proxied by another remote+ -- which is itself configured as a cluster gateway.+ | otherwise = case remoteAnnexProxiedBy (Remote.gitconfig r) of+ Just proxyuuid -> not $ null $ + concatMap (remoteAnnexClusterGateway . Remote.gitconfig) $+ filter (\p -> Remote.uuid p == proxyuuid) rs+ Nothing -> False++ proxyforcluster cu = do+ clusters <- getClusters+ if M.member cu (clusterUUIDs clusters)+ then return (Right (Left cu))+ else notconfigured++ notconfigured = M.lookup remoteuuid <$> uuidDescMap >>= \case+ Just desc -> return $ Left $ Just $+ "not configured to proxy for repository " ++ fromUUIDDesc desc+ Nothing -> return $ Left Nothing++mkProxyMethods :: ProxyMethods+mkProxyMethods = ProxyMethods+ { removedContent = \u k -> logChange k u InfoMissing+ , addedContent = \u k -> logChange k u InfoPresent+ }
+ Annex/SafeDropProof.hs view
@@ -0,0 +1,34 @@+{- git-annex safe drop proof+ -+ - Copyright 2014-2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE OverloadedStrings #-}++module Annex.SafeDropProof (+ SafeDropProof,+ safeDropProofEndTime,+ safeDropProofExpired,+ checkSafeDropProofEndTime,+) where++import Annex.Common+import Types.NumCopies++import Data.Time.Clock.POSIX++safeDropProofExpired :: Annex ()+safeDropProofExpired = do+ showNote "unsafe"+ showLongNote $ UnquotedString+ "Dropping took too long, and locks may have expired."++checkSafeDropProofEndTime :: Maybe SafeDropProof -> IO Bool+checkSafeDropProofEndTime p = case safeDropProofEndTime =<< p of+ Nothing -> return True+ Just endtime -> do+ now <- getPOSIXTime+ return (endtime > now)+
Annex/Verify.hs view
@@ -1,6 +1,6 @@ {- verification -- - Copyright 2010-2022 Joey Hess <id@joeyh.name>+ - Copyright 2010-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -19,8 +19,11 @@ isVerifiable, startVerifyKeyContentIncrementally, finishVerifyKeyContentIncrementally,+ finishVerifyKeyContentIncrementally', verifyKeyContentIncrementally, IncrementalVerifier(..),+ writeVerifyChunk,+ resumeVerifyFromOffset, tailVerify, ) where @@ -32,6 +35,7 @@ import qualified Backend import Types.Remote (unVerified, Verification(..), RetrievalSecurityPolicy(..)) import Utility.Hash (IncrementalVerifier(..))+import Utility.Metered import Annex.WorkerPool import Types.WorkerPool import Types.Key@@ -196,13 +200,17 @@ ) finishVerifyKeyContentIncrementally :: Maybe IncrementalVerifier -> Annex (Bool, Verification)-finishVerifyKeyContentIncrementally Nothing = +finishVerifyKeyContentIncrementally = finishVerifyKeyContentIncrementally' False++finishVerifyKeyContentIncrementally' :: Bool -> Maybe IncrementalVerifier -> Annex (Bool, Verification)+finishVerifyKeyContentIncrementally' _ Nothing = return (True, UnVerified)-finishVerifyKeyContentIncrementally (Just iv) =+finishVerifyKeyContentIncrementally' quiet (Just iv) = liftIO (finalizeIncrementalVerifier iv) >>= \case Just True -> return (True, Verified) Just False -> do- warning "verification of content failed"+ unless quiet $+ warning "verification of content failed" return (False, UnVerified) -- Incremental verification was not able to be done. Nothing -> return (True, UnVerified)@@ -212,6 +220,44 @@ miv <- startVerifyKeyContentIncrementally verifyconfig k a miv snd <$> finishVerifyKeyContentIncrementally miv++writeVerifyChunk :: Maybe IncrementalVerifier -> Handle -> S.ByteString -> IO ()+writeVerifyChunk (Just iv) h c = do+ S.hPut h c+ updateIncrementalVerifier iv c+writeVerifyChunk Nothing h c = S.hPut h c++{- Given a file handle that is open for reading (and likely also for writing),+ - and an offset, feeds the current content of the file up to the offset to+ - the IncrementalVerifier. Leaves the file seeked to the offset.+ - Returns the meter with the offset applied. -}+resumeVerifyFromOffset+ :: Integer+ -> Maybe IncrementalVerifier+ -> MeterUpdate+ -> Handle+ -> IO MeterUpdate+resumeVerifyFromOffset o incrementalverifier meterupdate h+ | o /= 0 = do+ maybe noop (`go` o) incrementalverifier+ -- Make sure the handle is seeked to the offset.+ -- (Reading the file probably left it there+ -- when that was done, but let's be sure.)+ hSeek h AbsoluteSeek o+ return offsetmeterupdate+ | otherwise = return meterupdate+ where+ offsetmeterupdate = offsetMeterUpdate meterupdate (toBytesProcessed o)+ go iv n+ | n == 0 = return ()+ | otherwise = do+ let c = if n > fromIntegral defaultChunkSize+ then defaultChunkSize+ else fromIntegral n+ b <- S.hGet h c+ updateIncrementalVerifier iv b+ unless (b == S.empty) $+ go iv (n - fromIntegral (S.length b)) -- | Runs a writer action that retrieves to a file. In another thread, -- reads the file as it grows, and feeds it to the incremental verifier.
Annex/WorkerPool.hs view
@@ -97,13 +97,11 @@ -- | Waits until there's an idle StartStage worker in the worker pool, -- removes it from the pool, and returns its state.------ If the worker pool is not already allocated, returns Nothing.-waitStartWorkerSlot :: TMVar (WorkerPool t) -> STM (Maybe (t, WorkerStage))+waitStartWorkerSlot :: TMVar (WorkerPool t) -> STM (t, WorkerStage) waitStartWorkerSlot tv = do pool <- takeTMVar tv v <- go pool- return $ Just (v, StartStage)+ return (v, StartStage) where go pool = case spareVals pool of [] -> retry
Assistant/Threads/Committer.hs view
@@ -290,19 +290,34 @@ refillChanges postponed returnWhen (null toadd) $ do+ (addedpointerfiles, toaddrest) <- partitionEithers+ <$> mapM checkpointerfile toadd (toaddannexed, toaddsmall) <- partitionEithers- <$> mapM checksmall toadd+ <$> mapM checksmall toaddrest addsmall toaddsmall addedannexed <- addaction toadd $ catMaybes <$> addannexed toaddannexed- return $ addedannexed ++ toaddsmall ++ otherchanges+ return $ addedannexed ++ toaddsmall ++ addedpointerfiles ++ otherchanges where (incomplete, otherchanges) = partition (\c -> isPendingAddChange c || isInProcessAddChange c) cs returnWhen c a | c = return otherchanges | otherwise = a-+ + checkpointerfile change = do+ let file = toRawFilePath $ changeFile change+ mk <- liftIO $ isPointerFile file+ case mk of+ Nothing -> return (Right change)+ Just key -> do+ mode <- liftIO $ catchMaybeIO $ fileMode <$> R.getFileStatus file+ liftAnnex $ stagePointerFile file mode =<< hashPointerFile key+ return $ Left $ Change+ (changeTime change)+ (changeFile change)+ (LinkChange (Just key))+ checksmall change | not annexdotfiles && dotfile f = return (Right change)
Assistant/Threads/Watcher.hs view
@@ -196,11 +196,8 @@ shouldRestage ds = scanComplete ds || forceRestage ds onAddFile :: Bool -> Handler-onAddFile symlinkssupported f fs = do- mk <- liftIO $ isPointerFile $ toRawFilePath f- case mk of- Nothing -> onAddFile' contentchanged addassociatedfile addlink samefilestatus symlinkssupported f fs- Just k -> addlink f k+onAddFile symlinkssupported f fs =+ onAddFile' contentchanged addassociatedfile addlink samefilestatus symlinkssupported f fs where addassociatedfile key file = Database.Keys.addAssociatedFile key
BuildFlags.hs view
@@ -52,6 +52,11 @@ #ifdef WITH_MAGICMIME , "MagicMime" #endif+#ifdef WITH_SERVANT+ , "Servant"+#else+#warning Building without servant, will not support annex+http urls or git-annex p2phttp.+#endif #ifdef WITH_BENCHMARK , "Benchmark" #endif
CHANGELOG view
@@ -1,3 +1,29 @@+git-annex (10.20240731) upstream; urgency=medium++ * New HTTP API that is equivilant to the P2P protocol.+ * New p2phttp command to serve the HTTP API.+ * annex+http and annex+https urls can be configured for+ remote.name.annexUrl to use the HTTP API to communicate with a server.+ This supports writable repositories, as well as accessing clusters+ and proxied remotes over HTTP.+ * When a http remote has annex.url set to an annex+http url in+ the git config file on the website, it will be copied into + remote.name.annexUrl the first time git-annex uses the remote.+ * assistant: Fix a race condition that could cause a pointer file to+ get ingested into the annex.+ * Avoid potential data loss in unlikely situations where git-annex-shell+ or git-annex remotedaemon is killed while locking a key to prevent its+ removal.+ * When proxying a download from a special remote, avoid unncessary hashing.+ * When proxying an upload to a special remote, verify the hash.+ * Propagate --force to git-annex transferrer.+ * Added a build flag for servant, enabling annex+http urls and + git-annex p2phttp.+ * Added a dependency on the haskell clock library.+ * Updated stack.yaml to nightly-2024-07-29.++ -- Joey Hess <id@joeyh.name> Wed, 31 Jul 2024 14:02:21 -0400+ git-annex (10.20240701) upstream; urgency=medium * git-annex remotes can now act as proxies that provide access to
CmdLine/Action.hs view
@@ -87,9 +87,8 @@ runconcurrent sizelimit Nothing = runnonconcurrent sizelimit runconcurrent sizelimit (Just tv) = - liftIO (atomically (waitStartWorkerSlot tv)) >>= maybe- (runnonconcurrent sizelimit)- (runconcurrent' sizelimit tv)+ liftIO (atomically (waitStartWorkerSlot tv))+ >>= runconcurrent' sizelimit tv runconcurrent' sizelimit tv (workerstrd, workerstage) = do aid <- liftIO $ async $ snd <$> Annex.run workerstrd
CmdLine/GitAnnex.hs view
@@ -1,6 +1,6 @@ {- git-annex main program -- - Copyright 2010-2022 Joey Hess <id@joeyh.name>+ - Copyright 2010-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -118,6 +118,9 @@ import qualified Command.Forget import qualified Command.OldKeys import qualified Command.P2P+#ifdef WITH_SERVANT+import qualified Command.P2PHttp+#endif import qualified Command.Proxy import qualified Command.DiffDriver import qualified Command.Smudge@@ -245,6 +248,9 @@ , Command.Forget.cmd , Command.OldKeys.cmd , Command.P2P.cmd+#ifdef WITH_SERVANT+ , Command.P2PHttp.cmd+#endif , Command.Proxy.cmd , Command.DiffDriver.cmd , Command.Smudge.cmd
CmdLine/GitAnnexShell.hs view
@@ -8,7 +8,6 @@ module CmdLine.GitAnnexShell where import Annex.Common-import qualified Annex import qualified Git.Construct import qualified Git.Config import CmdLine@@ -20,11 +19,7 @@ import Remote.GCrypt (getGCryptUUID) import P2P.Protocol (ServerMode(..)) import Git.Types-import qualified Types.Remote as R-import Logs.Proxy-import Logs.Cluster-import Logs.UUID-import Remote+import Annex.Proxy import qualified Command.ConfigList import qualified Command.NotifyChanges@@ -36,7 +31,6 @@ import qualified Command.DropKey import qualified Data.Map as M-import qualified Data.Set as S cmdsMap :: M.Map ServerMode [Command] cmdsMap = M.fromList $ map mk@@ -90,7 +84,7 @@ check u | u == toUUID expected = noop | otherwise = - unlessM (checkProxy (toUUID expected) u) $+ unlessM (checkCanProxy (toUUID expected) u) $ unexpectedUUID expected u checkGCryptUUID expected = check =<< getGCryptUUID True =<< gitRepo@@ -184,61 +178,3 @@ | field == fieldName remoteUUID = fieldCheck remoteUUID val | field == fieldName autoInit = fieldCheck autoInit val | otherwise = False--{- Check if this repository can proxy for a specified remote uuid,- - and if so enable proxying for it. -}-checkProxy :: UUID -> UUID -> Annex Bool-checkProxy remoteuuid ouruuid = M.lookup ouruuid <$> getProxies >>= \case- Nothing -> return False- -- This repository has (or had) proxying enabled. So it's- -- ok to display error messages that talk about proxies.- Just proxies ->- case filter (\p -> proxyRemoteUUID p == remoteuuid) (S.toList proxies) of- [] -> notconfigured- ps -> case mkClusterUUID remoteuuid of- Just cu -> proxyforcluster cu- Nothing -> proxyfor ps- where- -- This repository may have multiple remotes that access the same- -- repository. Proxy for the lowest cost one that is configured to- -- be used as a proxy.- proxyfor ps = do- rs <- concat . byCost <$> remoteList- myclusters <- annexClusters <$> Annex.getGitConfig- let sameuuid r = uuid r == remoteuuid- let samename r p = name r == proxyRemoteName p- case headMaybe (filter (\r -> sameuuid r && proxyisconfigured rs myclusters r && any (samename r) ps) rs) of- Nothing -> notconfigured- Just r -> do- Annex.changeState $ \st ->- st { Annex.proxyremote = Just (Right r) }- return True- - -- Only proxy for a remote when the git configuration- -- allows it. This is important to prevent changes to - -- the git-annex branch making git-annex-shell unexpectedly- -- proxy for remotes.- proxyisconfigured rs myclusters r- | remoteAnnexProxy (R.gitconfig r) = True- -- Proxy for remotes that are configured as cluster nodes.- | any (`M.member` myclusters) (fromMaybe [] $ remoteAnnexClusterNode $ R.gitconfig r) = True- -- Proxy for a remote when it is proxied by another remote- -- which is itself configured as a cluster gateway.- | otherwise = case remoteAnnexProxiedBy (R.gitconfig r) of- Just proxyuuid -> not $ null $ - concatMap (remoteAnnexClusterGateway . R.gitconfig) $- filter (\p -> R.uuid p == proxyuuid) rs- Nothing -> False-- proxyforcluster cu = do- clusters <- getClusters- if M.member cu (clusterUUIDs clusters)- then do- Annex.changeState $ \st ->- st { Annex.proxyremote = Just (Left cu) }- return True- else notconfigured-- notconfigured = M.lookup remoteuuid <$> uuidDescMap >>= \case- Just desc -> giveup $ "not configured to proxy for repository " ++ fromUUIDDesc desc- Nothing -> return False
CmdLine/GitRemoteAnnex.hs view
@@ -996,7 +996,7 @@ dropKey' :: Remote -> Key -> Annex () dropKey' rmt k = getKeyExportLocations rmt k >>= \case- Nothing -> Remote.removeKey rmt k+ Nothing -> Remote.removeKey rmt Nothing k Just locs -> forM_ locs $ \loc -> Remote.removeExport (Remote.exportActions rmt) k loc
Command/Drop.hs view
@@ -151,7 +151,7 @@ , "proof:" , show proof ]- ok <- Remote.action (Remote.removeKey remote key)+ ok <- Remote.action (Remote.removeKey remote proof key) next $ cleanupRemote key remote ud ok , stop )
Command/ExtendCluster.hs view
@@ -50,7 +50,7 @@ let setcus f = setConfig f (fromUUID (fromClusterUUID cu)) unless (M.member clustername myclusters) $ do setcus $ annexConfig ("cluster." <> encodeBS clustername)- setcus $ remoteAnnexConfig gatewayremote $ + setcus $ mkRemoteConfigKey gatewayremote $ remoteGitConfigKey ClusterGatewayField next $ return True where
Command/Fsck.hs view
@@ -639,7 +639,7 @@ ) ) - dropped <- tryNonAsync (Remote.removeKey remote key)+ dropped <- tryNonAsync (Remote.removeKey remote Nothing key) when (isRight dropped) $ Remote.logStatus remote key InfoMissing return $ case (movedbad, dropped) of
Command/Move.hs view
@@ -290,29 +290,32 @@ next $ return True -- copy complete finish deststartedwithcopy True RemoveSafe = do destuuid <- getUUID- lockContentShared key $ \_lck ->+ lockContentShared key Nothing $ \_lck -> fromDrop src destuuid deststartedwithcopy key afile id fromDrop :: Remote -> UUID -> DestStartedWithCopy -> Key -> AssociatedFile -> ([UnVerifiedCopy] -> [UnVerifiedCopy])-> CommandPerform fromDrop src destuuid deststartedwithcopy key afile adjusttocheck = willDropMakeItWorse (Remote.uuid src) destuuid deststartedwithcopy key afile >>= \case- DropAllowed -> dropremote "moved"+ DropAllowed -> dropremote Nothing "moved" DropCheckNumCopies -> do (numcopies, mincopies) <- getSafestNumMinCopies afile key (tocheck, verified) <- verifiableCopies key [Remote.uuid src] verifyEnoughCopiesToDrop "" key (Just (Remote.uuid src)) Nothing numcopies mincopies [Remote.uuid src] verified- (adjusttocheck tocheck) (dropremote . showproof) faileddropremote+ (adjusttocheck tocheck) dropremotewithproof faileddropremote DropWorse -> faileddropremote where showproof proof = "proof: " ++ show proof - dropremote reason = do+ dropremotewithproof proof = + dropremote (Just proof) (showproof proof)++ dropremote mproof reason = do fastDebug "Command.Move" $ unwords [ "Dropping from remote" , show src , "(" ++ reason ++ ")" ]- ok <- Remote.action (Remote.removeKey src key)+ ok <- Remote.action (Remote.removeKey src mproof key) when ok $ logMoveCleanup deststartedwithcopy next $ Command.Drop.cleanupRemote key src (Command.Drop.DroppingUnused False) ok
+ Command/P2PHttp.hs view
@@ -0,0 +1,174 @@+{- git-annex command+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE DataKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE OverloadedStrings #-}++module Command.P2PHttp where++import Command+import P2P.Http.Server+import P2P.Http.Url+import qualified P2P.Protocol as P2P+import Utility.Env++import Servant+import qualified Network.Wai.Handler.Warp as Warp+import qualified Network.Wai.Handler.WarpTLS as Warp+import Network.Socket (PortNumber)+import qualified Data.Map as M+import Data.String++cmd :: Command+cmd = noMessages $ withAnnexOptions [jobsOption] $+ command "p2phttp" SectionPlumbing+ "communicate in P2P protocol over http"+ paramNothing (seek <$$> optParser)++data Options = Options+ { portOption :: Maybe PortNumber+ , bindOption :: Maybe String+ , certFileOption :: Maybe FilePath+ , privateKeyFileOption :: Maybe FilePath+ , chainFileOption :: [FilePath]+ , authEnvOption :: Bool+ , authEnvHttpOption :: Bool+ , unauthReadOnlyOption :: Bool+ , unauthAppendOnlyOption :: Bool+ , wideOpenOption :: Bool+ , proxyConnectionsOption :: Maybe Integer+ , clusterJobsOption :: Maybe Int+ }++optParser :: CmdParamsDesc -> Parser Options+optParser _ = Options+ <$> optional (option auto+ ( long "port" <> metavar paramNumber+ <> help "specify port to listen on"+ ))+ <*> optional (strOption+ ( long "bind" <> metavar paramAddress+ <> help "specify address to bind to"+ ))+ <*> optional (strOption+ ( long "certfile" <> metavar paramFile+ <> help "TLS certificate file for HTTPS"+ ))+ <*> optional (strOption+ ( long "privatekeyfile" <> metavar paramFile+ <> help "TLS private key file for HTTPS"+ ))+ <*> many (strOption+ ( long "chainfile" <> metavar paramFile+ <> help "TLS chain file"+ ))+ <*> switch+ ( long "authenv"+ <> help "authenticate users from environment (https only)"+ )+ <*> switch+ ( long "authenv-http"+ <> help "authenticate users from environment (including http)"+ )+ <*> switch+ ( long "unauth-readonly"+ <> help "allow unauthenticated users to read the repository"+ )+ <*> switch+ ( long "unauth-appendonly"+ <> help "allow unauthenticated users to read and append to the repository"+ )+ <*> switch+ ( long "wideopen"+ <> help "give unauthenticated users full read+write access"+ )+ <*> optional (option auto+ ( long "proxyconnections" <> metavar paramNumber+ <> help "maximum number of idle connections when proxying"+ ))+ <*> optional (option auto+ ( long "clusterjobs" <> metavar paramNumber+ <> help "number of concurrent node accesses per connection"+ ))++seek :: Options -> CommandSeek+seek o = getAnnexWorkerPool $ \workerpool ->+ withP2PConnections workerpool+ (fromMaybe 1 $ proxyConnectionsOption o)+ (fromMaybe 1 $ clusterJobsOption o)+ (go workerpool)+ where+ go workerpool acquireconn = liftIO $ do+ authenv <- getAuthEnv+ st <- mkP2PHttpServerState acquireconn workerpool $+ mkGetServerMode authenv o+ let settings = Warp.setPort port $ Warp.setHost host $+ Warp.defaultSettings+ case (certFileOption o, privateKeyFileOption o) of+ (Nothing, Nothing) -> Warp.runSettings settings (p2pHttpApp st)+ (Just certfile, Just privatekeyfile) -> do+ let tlssettings = Warp.tlsSettingsChain+ certfile (chainFileOption o) privatekeyfile+ Warp.runTLS tlssettings settings (p2pHttpApp st)+ _ -> giveup "You must use both --certfile and --privatekeyfile options to enable HTTPS."+ + port = maybe+ (fromIntegral defaultP2PHttpProtocolPort)+ fromIntegral+ (portOption o)+ host = maybe+ (fromString "*") -- both ipv4 and ipv6+ fromString+ (bindOption o)++mkGetServerMode :: M.Map Auth P2P.ServerMode -> Options -> GetServerMode+mkGetServerMode _ o _ Nothing+ | wideOpenOption o = Just P2P.ServeReadWrite+ | unauthAppendOnlyOption o = Just P2P.ServeAppendOnly+ | unauthReadOnlyOption o = Just P2P.ServeReadOnly+ | otherwise = Nothing+mkGetServerMode authenv o issecure (Just auth) =+ case (issecure, authEnvOption o, authEnvHttpOption o) of+ (Secure, True, _) -> checkauth+ (NotSecure, _, True) -> checkauth+ _ -> noauth+ where+ checkauth = case M.lookup auth authenv of+ Just servermode -> Just servermode+ Nothing -> noauth+ noauth = mkGetServerMode authenv o issecure Nothing++getAuthEnv :: IO (M.Map Auth P2P.ServerMode)+getAuthEnv = do+ environ <- getEnvironment+ let permmap = M.fromList (mapMaybe parseperms environ)+ return $ M.fromList $+ map (addperms permmap) $+ mapMaybe parseusername environ+ where+ parseperms (k, v) = case deprefix "GIT_ANNEX_P2PHTTP_PERMISSIONS_" k of+ Nothing -> Nothing+ Just username -> case v of+ "readonly" -> Just+ (encodeBS username, P2P.ServeReadOnly)+ "appendonly" -> Just+ (encodeBS username, P2P.ServeAppendOnly)+ _ -> Nothing++ parseusername (k, v) = case deprefix "GIT_ANNEX_P2PHTTP_PASSWORD_" k of+ Nothing -> Nothing+ Just username -> Just $ Auth (encodeBS username) (encodeBS v)++ deprefix prefix s+ | prefix `isPrefixOf` s = Just (drop (length prefix) s)+ | otherwise = Nothing++ addperms permmap auth@(Auth user _) = + case M.lookup user permmap of+ Nothing -> (auth, P2P.ServeReadWrite)+ Just perms -> (auth, perms)
Command/P2PStdIO.hs view
@@ -16,7 +16,6 @@ import Annex.Proxy import Annex.UUID import qualified CmdLine.GitAnnexShell.Checks as Checks-import Logs.Location import Logs.Cluster import Annex.Cluster import qualified Remote@@ -61,7 +60,7 @@ performProxy :: UUID -> P2P.ServerMode -> Remote -> CommandPerform performProxy clientuuid servermode r = do- clientside <- proxyClientSide clientuuid+ clientside <- mkProxyClientSide clientuuid getClientProtocolVersion (Remote.uuid r) clientside (withclientversion clientside) (p2pErrHandler noop)@@ -76,28 +75,30 @@ closeRemoteSide remoteside p2pDone let errhandler = p2pErrHandler (closeRemoteSide remoteside)- let runproxy othermsg' = proxy closer proxymethods- servermode clientside- (Remote.uuid r)- (singleProxySelector remoteside)- concurrencyconfig- protocolversion othermsg' errhandler+ proxystate <- liftIO mkProxyState+ let proxyparams = ProxyParams+ { proxyMethods = mkProxyMethods+ , proxyState = proxystate+ , proxyServerMode = servermode+ , proxyClientSide = clientside+ , proxyUUID = Remote.uuid r+ , proxySelector = singleProxySelector remoteside+ , proxyConcurrencyConfig = concurrencyconfig+ , proxyClientProtocolVersion = protocolversion+ }+ let runproxy othermsg' = proxy closer proxyparams + othermsg' errhandler sendClientProtocolVersion clientside othermsg protocolversion runproxy errhandler withclientversion _ Nothing = p2pDone- - proxymethods = ProxyMethods- { removedContent = \u k -> logChange k u InfoMissing- , addedContent = \u k -> logChange k u InfoPresent- } performProxyCluster :: UUID -> ClusterUUID -> P2P.ServerMode -> CommandPerform performProxyCluster clientuuid clusteruuid servermode = do- clientside <- proxyClientSide clientuuid+ clientside <- mkProxyClientSide clientuuid proxyCluster clusteruuid p2pDone servermode clientside p2pErrHandler -proxyClientSide :: UUID -> Annex ClientSide-proxyClientSide clientuuid = do+mkProxyClientSide :: UUID -> Annex ClientSide+mkProxyClientSide clientuuid = do clientrunst <- liftIO (mkRunState $ Serving clientuuid Nothing) ClientSide clientrunst <$> liftIO (stdioP2PConnectionDupped Nothing)
Command/TestRemote.hs view
@@ -303,7 +303,7 @@ Right v -> return (True, v) Left _ -> return (False, UnVerified) store r k = Remote.storeKey r k (AssociatedFile Nothing) Nothing nullMeterUpdate- remove r k = Remote.removeKey r k+ remove r k = Remote.removeKey r Nothing k testExportTree :: RunAnnex -> Annex (Maybe Remote) -> Annex Key -> Annex Key -> [TestTree] testExportTree runannex mkr mkk1 mkk2 =@@ -366,7 +366,7 @@ testUnavailable :: RunAnnex -> Annex (Maybe Remote) -> Annex Key -> [TestTree] testUnavailable runannex mkr mkk = [ check isLeft "removeKey" $ \r k ->- Remote.removeKey r k+ Remote.removeKey r Nothing k , check isLeft "storeKey" $ \r k -> Remote.storeKey r k (AssociatedFile Nothing) Nothing nullMeterUpdate , check (`notElem` [Right True, Right False]) "checkPresent" $ \r k ->@@ -397,7 +397,7 @@ cleanup rs ks ok | all Remote.readonly rs = return ok | otherwise = do- forM_ rs $ \r -> forM_ ks (Remote.removeKey r)+ forM_ rs $ \r -> forM_ ks (Remote.removeKey r Nothing) forM_ ks $ \k -> lockContentForRemoval k noop removeAnnex return ok
Git/Credential.hs view
@@ -42,7 +42,7 @@ Just c -> go (const noop) c Nothing -> do let storeincache = \c -> atomically $ do- (CredentialCache cc') <- takeTMVar ccv+ CredentialCache cc' <- takeTMVar ccv putTMVar ccv (CredentialCache (M.insert bu c cc')) go storeincache =<< getUrlCredential u r Nothing -> go (const noop) =<< getUrlCredential u r@@ -113,7 +113,9 @@ -- when credential.useHttpPath is false, one Credential is cached -- for each git repo accessed, and there are a reasonably small number of -- those, so the cache will not grow too large.-data CredentialBaseURL = CredentialBaseURL URI+data CredentialBaseURL+ = CredentialBaseURI URI+ | CredentialBaseURL String deriving (Show, Eq, Ord) mkCredentialBaseURL :: Repo -> URLString -> Maybe CredentialBaseURL@@ -123,4 +125,4 @@ Config.get (ConfigKey "credential.useHttpPath") (ConfigValue "") r if usehttppath then Nothing- else Just $ CredentialBaseURL $ u { uriPath = "" }+ else Just $ CredentialBaseURI $ u { uriPath = "" }
P2P/Annex.hs view
@@ -1,6 +1,6 @@ {- P2P protocol, Annex implementation -- - Copyright 2016-2022 Joey Hess <id@joeyh.name>+ - Copyright 2016-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -23,11 +23,12 @@ import Logs.Location import Types.NumCopies import Utility.Metered+import Utility.MonotonicClock import Annex.Verify import Control.Monad.Free import Control.Concurrent.STM-import qualified Data.ByteString as S+import Data.Time.Clock.POSIX -- Full interpreter for Proto, that can receive and send objects. runFullProto :: RunState -> P2PConnection -> Proto a -> Annex (Either ProtoFailure a)@@ -99,6 +100,26 @@ Left e -> return $ Left $ ProtoFailureException e Right (Left e) -> return $ Left e Right (Right ok) -> runner (next ok)+ SendContentWith consumer getb validitycheck next -> do+ v <- tryNonAsync $ do+ let fallback = return $ Left $+ ProtoFailureMessage "Transfer failed"+ let consumer' b ti = do+ validator <- consumer b+ indicatetransferred ti+ return validator+ runner getb >>= \case+ Left e -> giveup $ describeProtoFailure e+ Right b -> checktransfer (\ti -> Right <$> consumer' b ti) fallback >>= \case+ Left e -> return (Left e)+ Right validator ->+ runner validitycheck >>= \case+ Right v -> Right <$> validator v+ _ -> Right <$> validator Nothing+ case v of+ Left e -> return $ Left $ ProtoFailureException e+ Right (Left e) -> return $ Left e+ Right (Right ok) -> runner (next ok) SetPresent k u next -> do v <- tryNonAsync $ logChange k u InfoPresent case v of@@ -109,22 +130,31 @@ case v of Left e -> return $ Left $ ProtoFailureException e Right result -> runner (next result)- RemoveContent k next -> do+ RemoveContent k mts next -> do let cleanup = do logStatus k InfoMissing return True+ let checkts = case mts of+ Nothing -> return True+ Just ts -> do+ now <- liftIO currentMonotonicTimestamp+ return (now < ts) v <- tryNonAsync $ ifM (Annex.Content.inAnnex k)- ( lockContentForRemoval k cleanup $ \contentlock -> do- removeAnnex contentlock- cleanup+ ( lockContentForRemoval k cleanup $ \contentlock ->+ ifM checkts+ ( do+ removeAnnex contentlock+ cleanup+ , return False+ ) , return True ) case v of Left e -> return $ Left $ ProtoFailureException e Right result -> runner (next result) TryLockContent k protoaction next -> do- v <- tryNonAsync $ lockContentShared k $ \verifiedcopy -> + v <- tryNonAsync $ lockContentShared k (Just p2pDefaultLockContentRetentionDuration) $ \verifiedcopy -> case verifiedcopy of LockedCopy _ -> runner (protoaction True) _ -> runner (protoaction False)@@ -146,7 +176,10 @@ UpdateMeterTotalSize m sz next -> do liftIO $ setMeterTotalSize m sz runner next- RunValidityCheck checkaction next -> runner . next =<< checkaction+ RunValidityCheck checkaction next ->+ runner . next =<< checkaction+ GetLocalCurrentTime next ->+ runner . next =<< liftIO getPOSIXTime where transfer mk k af sd ta = case runst of -- Update transfer logs when serving.@@ -157,43 +190,13 @@ -- a client. Client _ -> ta nullMeterUpdate - resumefromoffset o incrementalverifier p h- | o /= 0 = do- p' <- case incrementalverifier of- Just iv -> do- go iv o- return p- _ -> return $ offsetMeterUpdate p (toBytesProcessed o)- -- Make sure the handle is seeked to the offset.- -- (Reading the file probably left it there- -- when that was done, but let's be sure.)- hSeek h AbsoluteSeek o- return p'- | otherwise = return p- where- go iv n- | n == 0 = return ()- | otherwise = do- let c = if n > fromIntegral defaultChunkSize- then defaultChunkSize- else fromIntegral n- b <- S.hGet h c- updateIncrementalVerifier iv b- unless (b == S.empty) $- go iv (n - fromIntegral (S.length b))- storefile dest (Offset o) (Len l) getb incrementalverifier validitycheck p ti = do v <- runner getb case v of Right b -> do liftIO $ withBinaryFile dest ReadWriteMode $ \h -> do- p' <- resumefromoffset o incrementalverifier p h- let writechunk = case incrementalverifier of- Nothing -> \c -> S.hPut h c- Just iv -> \c -> do- S.hPut h c- updateIncrementalVerifier iv c- meteredWrite p' writechunk b+ p' <- resumeVerifyFromOffset o incrementalverifier p h+ meteredWrite p' (writeVerifyChunk incrementalverifier h) b indicatetransferred ti rightsize <- do
+ P2P/Http.hs view
@@ -0,0 +1,184 @@+{- P2P protocol over HTTP+ -+ - https://git-annex.branchable.com/design/p2p_protocol_over_http/+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE DataKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-}++module P2P.Http (+ module P2P.Http,+ module P2P.Http.Types,+) where++import P2P.Http.Types++import Servant+import qualified Data.ByteString as B++type P2PHttpAPI+ = "git-annex" :> SU :> PV3 :> "key" :> GetAPI+ :<|> "git-annex" :> SU :> PV2 :> "key" :> GetAPI+ :<|> "git-annex" :> SU :> PV1 :> "key" :> GetAPI+ :<|> "git-annex" :> SU :> PV0 :> "key" :> GetAPI+ :<|> "git-annex" :> SU :> PV3 :> "checkpresent" :> CheckPresentAPI+ :<|> "git-annex" :> SU :> PV2 :> "checkpresent" :> CheckPresentAPI+ :<|> "git-annex" :> SU :> PV1 :> "checkpresent" :> CheckPresentAPI+ :<|> "git-annex" :> SU :> PV0 :> "checkpresent" :> CheckPresentAPI+ :<|> "git-annex" :> SU :> PV3 :> "remove" :> RemoveAPI RemoveResultPlus+ :<|> "git-annex" :> SU :> PV2 :> "remove" :> RemoveAPI RemoveResultPlus+ :<|> "git-annex" :> SU :> PV1 :> "remove" :> RemoveAPI RemoveResult+ :<|> "git-annex" :> SU :> PV0 :> "remove" :> RemoveAPI RemoveResult+ :<|> "git-annex" :> SU :> PV3 :> "remove-before" :> RemoveBeforeAPI+ :<|> "git-annex" :> SU :> PV3 :> "gettimestamp" :> GetTimestampAPI+ :<|> "git-annex" :> SU :> PV3 :> "put" :> PutAPI PutResultPlus+ :<|> "git-annex" :> SU :> PV2 :> "put" :> PutAPI PutResultPlus+ :<|> "git-annex" :> SU :> PV1 :> "put" :> PutAPI PutResult+ :<|> "git-annex" :> SU :> PV0 :> "put" :> PutAPI PutResult+ :<|> "git-annex" :> SU :> PV3 :> "putoffset"+ :> PutOffsetAPI PutOffsetResultPlus+ :<|> "git-annex" :> SU :> PV2 :> "putoffset"+ :> PutOffsetAPI PutOffsetResultPlus+ :<|> "git-annex" :> SU :> PV1 :> "putoffset"+ :> PutOffsetAPI PutOffsetResult+ :<|> "git-annex" :> SU :> PV3 :> "lockcontent" :> LockContentAPI+ :<|> "git-annex" :> SU :> PV2 :> "lockcontent" :> LockContentAPI+ :<|> "git-annex" :> SU :> PV1 :> "lockcontent" :> LockContentAPI+ :<|> "git-annex" :> SU :> PV0 :> "lockcontent" :> LockContentAPI+ :<|> "git-annex" :> SU :> PV3 :> "keeplocked" :> KeepLockedAPI+ :<|> "git-annex" :> SU :> PV2 :> "keeplocked" :> KeepLockedAPI+ :<|> "git-annex" :> SU :> PV1 :> "keeplocked" :> KeepLockedAPI+ :<|> "git-annex" :> SU :> PV0 :> "keeplocked" :> KeepLockedAPI+ :<|> "git-annex" :> SU :> "key" :> GetGenericAPI++p2pHttpAPI :: Proxy P2PHttpAPI+p2pHttpAPI = Proxy++type GetGenericAPI+ = CaptureKey+ :> CU Optional+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> StreamGet NoFraming OctetStream+ (Headers '[DataLengthHeader] (SourceIO B.ByteString))++type GetAPI+ = CaptureKey+ :> CU Required+ :> BypassUUIDs+ :> AssociatedFileParam+ :> OffsetParam+ :> IsSecure+ :> AuthHeader+ :> StreamGet NoFraming OctetStream+ (Headers '[DataLengthHeader] (SourceIO B.ByteString))++type CheckPresentAPI+ = KeyParam+ :> CU Required+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] CheckPresentResult++type RemoveAPI result+ = KeyParam+ :> CU Required+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] result+ +type RemoveBeforeAPI+ = KeyParam+ :> CU Required+ :> BypassUUIDs+ :> QueryParam' '[Required] "timestamp" Timestamp+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] RemoveResultPlus++type GetTimestampAPI+ = CU Required+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] GetTimestampResult++type PutAPI result+ = DataLengthHeaderRequired+ :> KeyParam+ :> CU Required+ :> BypassUUIDs+ :> AssociatedFileParam+ :> OffsetParam+ :> StreamBody NoFraming OctetStream (SourceIO B.ByteString)+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] result++type PutOffsetAPI result+ = KeyParam+ :> CU Required+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] result++type LockContentAPI+ = KeyParam+ :> CU Required+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Post '[JSON] LockResult++type KeepLockedAPI+ = LockIDParam+ :> CU Optional+ :> BypassUUIDs+ :> IsSecure+ :> AuthHeader+ :> Header "Connection" ConnectionKeepAlive+ :> Header "Keep-Alive" KeepAlive+ :> StreamBody NewlineFraming JSON (SourceIO UnlockRequest)+ :> Post '[JSON] LockResult++type SU = Capture "serveruuid" (B64UUID ServerSide)++type CU req = QueryParam' '[req] "clientuuid" (B64UUID ClientSide)++type BypassUUIDs = QueryParams "bypass" (B64UUID Bypass)++type CaptureKey = Capture "key" B64Key++type KeyParam = QueryParam' '[Required] "key" B64Key++type AssociatedFileParam = QueryParam "associatedfile" B64FilePath++type OffsetParam = QueryParam "offset" Offset++type DataLengthHeader = Header DataLengthHeader' DataLength++type DataLengthHeaderRequired = Header' '[Required] DataLengthHeader' DataLength++type DataLengthHeader' = "X-git-annex-data-length"++type LockIDParam = QueryParam' '[Required] "lockid" LockID++type AuthHeader = Header "Authorization" Auth++type PV3 = Capture "v3" V3+type PV2 = Capture "v2" V2+type PV1 = Capture "v1" V1+type PV0 = Capture "v0" V0+
+ P2P/Http/Client.hs view
@@ -0,0 +1,535 @@+{- P2P protocol over HTTP, client+ -+ - https://git-annex.branchable.com/design/p2p_protocol_over_http/+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DataKinds, TypeApplications #-}+{-# LANGUAGE CPP #-}++module P2P.Http.Client (+ module P2P.Http.Client,+ module P2P.Http.Types,+ Validity(..),+) where++import Types+import P2P.Http.Types+import P2P.Protocol hiding (Offset, Bypass, auth, FileSize)+import Utility.Metered+import Utility.FileSize+import Types.NumCopies++import Annex.Common+#ifdef WITH_SERVANT+import qualified Annex+import Annex.UUID+import Annex.Url+import Types.Remote+import P2P.Http+import P2P.Http.Url+import Annex.Concurrent+import Utility.Url (BasicAuth(..))+import Utility.HumanTime+import qualified Git.Credential as Git++import Servant hiding (BasicAuthData(..))+import Servant.Client.Streaming+import qualified Servant.Types.SourceT as S+import Network.HTTP.Types.Status+import Network.HTTP.Client+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy.Internal as LI+import qualified Data.Map as M+import Control.Concurrent.STM+import Control.Concurrent.Async+import Control.Concurrent+import System.IO.Unsafe+#endif+import Data.Time.Clock.POSIX+import qualified Data.ByteString.Lazy as L++type ClientAction a+#ifdef WITH_SERVANT+ = ClientEnv+ -> ProtocolVersion+ -> B64UUID ServerSide+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> Maybe Auth+ -> Annex (Either ClientError a)+#else+ = ()+#endif++p2pHttpClient+ :: Remote+ -> (String -> Annex a)+ -> ClientAction a+ -> Annex a+p2pHttpClient rmt fallback clientaction = + p2pHttpClientVersions (const True) rmt fallback clientaction >>= \case+ Just res -> return res+ Nothing -> fallback "git-annex HTTP API server is missing an endpoint"++p2pHttpClientVersions+ :: (ProtocolVersion -> Bool)+ -> Remote+ -> (String -> Annex a)+ -> ClientAction a+ -> Annex (Maybe a)+#ifdef WITH_SERVANT+p2pHttpClientVersions allowedversion rmt fallback clientaction =+ case p2pHttpBaseUrl <$> remoteAnnexP2PHttpUrl (gitconfig rmt) of+ Nothing -> error "internal"+ Just baseurl -> do+ mgr <- httpManager <$> getUrlOptions+ let clientenv = mkClientEnv mgr baseurl+ ccv <- Annex.getRead Annex.gitcredentialcache+ Git.CredentialCache cc <- liftIO $ atomically $+ readTMVar ccv+ case M.lookup (Git.CredentialBaseURL credentialbaseurl) cc of+ Nothing -> go clientenv Nothing False Nothing versions+ Just cred -> go clientenv (Just cred) True (credauth cred) versions+ where+ versions = filter allowedversion allProtocolVersions+ go clientenv mcred credcached mauth (v:vs) = do+ myuuid <- getUUID+ res <- clientaction clientenv v+ (B64UUID (uuid rmt))+ (B64UUID myuuid)+ []+ mauth+ case res of+ Right resp -> do+ unless credcached $ cachecred mcred+ return (Just resp)+ Left (FailureResponse _ resp)+ | statusCode (responseStatusCode resp) == 404 && not (null vs) ->+ go clientenv mcred credcached mauth vs+ | statusCode (responseStatusCode resp) == 401 ->+ case mcred of+ Nothing -> authrequired clientenv (v:vs)+ Just cred -> do+ inRepo $ Git.rejectUrlCredential cred+ Just <$> fallback (showstatuscode resp)+ | otherwise -> Just <$> fallback (showstatuscode resp)+ Left (ConnectionError ex) -> case fromException ex of+ Just (HttpExceptionRequest _ (ConnectionFailure err)) -> Just <$> fallback+ ("unable to connect to HTTP server: " ++ show err)+ _ -> Just <$> fallback (show ex)+ Left clienterror -> Just <$> fallback+ ("git-annex HTTP API server returned an unexpected response: " ++ show clienterror)+ go _ _ _ _ [] = return Nothing++ authrequired clientenv vs = do+ cred <- prompt $ + inRepo $ Git.getUrlCredential credentialbaseurl+ go clientenv (Just cred) False (credauth cred) vs++ showstatuscode resp = + show (statusCode (responseStatusCode resp))+ ++ " " +++ decodeBS (statusMessage (responseStatusCode resp))++ credentialbaseurl = case p2pHttpUrlString <$> remoteAnnexP2PHttpUrl (gitconfig rmt) of+ Nothing -> error "internal"+ Just url -> url++ credauth cred = do+ ba <- Git.credentialBasicAuth cred+ return $ Auth+ (encodeBS (basicAuthUser ba))+ (encodeBS (basicAuthPassword ba))+ + cachecred mcred = case mcred of+ Just cred -> do+ inRepo $ Git.approveUrlCredential cred+ ccv <- Annex.getRead Annex.gitcredentialcache+ liftIO $ atomically $ do+ Git.CredentialCache cc <- takeTMVar ccv+ putTMVar ccv $ Git.CredentialCache $+ M.insert (Git.CredentialBaseURL credentialbaseurl) cred cc+ Nothing -> noop+#else+p2pHttpClientVersions _ _ fallback () = Just <$> fallback+ "This remote uses an annex+http url, but this version of git-annex is not built with support for that."+#endif++clientGet+ :: Key+ -> AssociatedFile+ -> (L.ByteString -> IO BytesProcessed)+ -- ^ Must consume the entire ByteString before returning its+ -- total size.+ -> Maybe FileSize+ -- ^ Size of existing file, when resuming.+ -> ClientAction Validity+#ifdef WITH_SERVANT+clientGet k af consumer startsz clientenv (ProtocolVersion ver) su cu bypass auth = liftIO $ do+ let offset = fmap (Offset . fromIntegral) startsz+ withClientM (cli (B64Key k) cu bypass baf offset auth) clientenv $ \case+ Left err -> return (Left err)+ Right respheaders -> do+ b <- S.unSourceT (getResponse respheaders) gather+ BytesProcessed len <- consumer b+ let DataLength dl = case lookupResponseHeader @DataLengthHeader' respheaders of+ Header hdr -> hdr+ _ -> error "missing data length header"+ return $ Right $ + if dl == len then Valid else Invalid+ where+ cli =case ver of+ 3 -> v3 su V3+ 2 -> v2 su V2+ 1 -> v1 su V1+ 0 -> v0 su V0+ _ -> error "unsupported protocol version"+ + v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI++ gather = unsafeInterleaveIO . gather'+ gather' S.Stop = return LI.Empty+ gather' (S.Error err) = giveup err+ gather' (S.Skip s) = gather' s+ gather' (S.Effect ms) = ms >>= gather'+ gather' (S.Yield v s) = LI.Chunk v <$> unsafeInterleaveIO (gather' s)+ + baf = associatedFileToB64FilePath af+#else+clientGet _ _ _ _ = ()+#endif++clientCheckPresent :: Key -> ClientAction Bool+#ifdef WITH_SERVANT+clientCheckPresent key clientenv (ProtocolVersion ver) su cu bypass auth =+ liftIO $ withClientM (cli su (B64Key key) cu bypass auth) clientenv $ \case+ Left err -> return (Left err)+ Right (CheckPresentResult res) -> return (Right res)+ where+ cli = case ver of+ 3 -> flip v3 V3+ 2 -> flip v2 V2+ 1 -> flip v1 V1+ 0 -> flip v0 V0+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI+#else+clientCheckPresent _ = ()+#endif++-- Similar to P2P.Protocol.remove.+clientRemoveWithProof+ :: Maybe SafeDropProof+ -> Key+ -> Annex RemoveResultPlus+ -> Remote+ -> Annex RemoveResultPlus+clientRemoveWithProof proof k unabletoremove remote =+ case safeDropProofEndTime =<< proof of+ Nothing -> removeanytime+ Just endtime -> removebefore endtime+ where+ removeanytime = p2pHttpClient remote giveup (clientRemove k)++ removebefore endtime =+ p2pHttpClientVersions useversion remote giveup clientGetTimestamp >>= \case+ Just (GetTimestampResult (Timestamp remotetime)) ->+ removebefore' endtime remotetime+ -- Peer is too old to support REMOVE-BEFORE.+ Nothing -> removeanytime+ + removebefore' endtime remotetime =+ canRemoveBefore endtime remotetime (liftIO getPOSIXTime) >>= \case+ Just remoteendtime -> p2pHttpClient remote giveup $+ clientRemoveBefore k (Timestamp remoteendtime)+ Nothing -> unabletoremove+ + useversion v = v >= ProtocolVersion 3++clientRemove :: Key -> ClientAction RemoveResultPlus+#ifdef WITH_SERVANT+clientRemove k clientenv (ProtocolVersion ver) su cu bypass auth =+ liftIO $ withClientM cli clientenv return+ where+ bk = B64Key k++ cli = case ver of+ 3 -> v3 su V3 bk cu bypass auth+ 2 -> v2 su V2 bk cu bypass auth+ 1 -> plus <$> v1 su V1 bk cu bypass auth+ 0 -> plus <$> v0 su V0 bk cu bypass auth+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI+#else+clientRemove _ = ()+#endif++clientRemoveBefore+ :: Key+ -> Timestamp+ -> ClientAction RemoveResultPlus+#ifdef WITH_SERVANT+clientRemoveBefore k ts clientenv (ProtocolVersion ver) su cu bypass auth =+ liftIO $ withClientM (cli su (B64Key k) cu bypass ts auth) clientenv return+ where+ cli = case ver of+ 3 -> flip v3 V3+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ v3 :<|> _ = client p2pHttpAPI+#else+clientRemoveBefore _ _ = ()+#endif++clientGetTimestamp :: ClientAction GetTimestampResult+#ifdef WITH_SERVANT+clientGetTimestamp clientenv (ProtocolVersion ver) su cu bypass auth = + liftIO $ withClientM (cli su cu bypass auth) clientenv return+ where+ cli = case ver of+ 3 -> flip v3 V3+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|>+ v3 :<|> _ = client p2pHttpAPI+#else+clientGetTimestamp = ()+#endif++clientPut+ :: MeterUpdate+ -> Key+ -> Maybe Offset+ -> AssociatedFile+ -> FilePath+ -> FileSize+ -> Annex Bool+ -- ^ Called after sending the file to check if it's valid.+ -> ClientAction PutResultPlus+#ifdef WITH_SERVANT+clientPut meterupdate k moffset af contentfile contentfilesize validitycheck clientenv (ProtocolVersion ver) su cu bypass auth = do+ checkv <- liftIO newEmptyTMVarIO+ checkresultv <- liftIO newEmptyTMVarIO+ let checker = do+ liftIO $ atomically $ takeTMVar checkv+ validitycheck >>= liftIO . atomically . putTMVar checkresultv+ checkerthread <- liftIO . async =<< forkState checker+ v <- liftIO $ withBinaryFile contentfile ReadMode $ \h -> do+ when (offset /= 0) $+ hSeek h AbsoluteSeek offset+ withClientM (cli (stream h checkv checkresultv)) clientenv return+ case v of+ Left err -> do+ void $ liftIO $ atomically $ tryPutTMVar checkv ()+ join $ liftIO (wait checkerthread)+ return (Left err)+ Right res -> do+ join $ liftIO (wait checkerthread)+ return (Right res)+ where+ stream h checkv checkresultv = S.SourceT $ \a -> do+ bl <- hGetContentsMetered h meterupdate+ v <- newMVar (0, filter (not . B.null) (L.toChunks bl))+ a (go v)+ where+ go v = S.fromActionStep B.null $ modifyMVar v $ \case+ (n, (b:[])) -> do+ let !n' = n + B.length b+ ifM (checkvalid n')+ ( return ((n', []), b)+ -- The key's content is invalid, but+ -- the amount of data is the same as+ -- the DataLengthHeader indicates.+ -- Truncate the stream by one byte to+ -- indicate to the server that it's+ -- not valid.+ , return + ( (n' - 1, [])+ , B.take (B.length b - 1) b+ )+ )+ (n, []) -> do+ void $ checkvalid n+ return ((n, []), mempty)+ (n, (b:bs)) ->+ let !n' = n + B.length b+ in return ((n', bs), b)++ checkvalid n = do+ void $ liftIO $ atomically $ tryPutTMVar checkv ()+ valid <- liftIO $ atomically $ readTMVar checkresultv+ if not valid+ then return (n /= fromIntegral nlen)+ else return True++ baf = case af of+ AssociatedFile Nothing -> Nothing+ AssociatedFile (Just f) -> Just (B64FilePath f)++ len = DataLength nlen++ nlen = contentfilesize - offset++ offset = case moffset of+ Nothing -> 0+ Just (Offset o) -> fromIntegral o+ + bk = B64Key k++ cli src = case ver of+ 3 -> v3 su V3 len bk cu bypass baf moffset src auth+ 2 -> v2 su V2 len bk cu bypass baf moffset src auth+ 1 -> plus <$> v1 su V1 len bk cu bypass baf moffset src auth+ 0 -> plus <$> v0 su V0 len bk cu bypass baf moffset src auth+ _ -> error "unsupported protocol version"++ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|>+ _ :<|>+ v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI+#else+clientPut _ _ _ _ _ _ _ = ()+#endif++clientPutOffset+ :: Key+ -> ClientAction PutOffsetResultPlus+#ifdef WITH_SERVANT+clientPutOffset k clientenv (ProtocolVersion ver) su cu bypass auth+ | ver == 0 = return (Right (PutOffsetResultPlus (Offset 0)))+ | otherwise = liftIO $ withClientM cli clientenv return+ where+ bk = B64Key k++ cli = case ver of+ 3 -> v3 su V3 bk cu bypass auth+ 2 -> v2 su V2 bk cu bypass auth+ 1 -> plus <$> v1 su V1 bk cu bypass auth+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|>+ _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ v3 :<|> v2 :<|> v1 :<|> _ = client p2pHttpAPI+#else+clientPutOffset _ = ()+#endif++clientLockContent+ :: Key+ -> ClientAction LockResult+#ifdef WITH_SERVANT+clientLockContent k clientenv (ProtocolVersion ver) su cu bypass auth = + liftIO $ withClientM (cli (B64Key k) cu bypass auth) clientenv return+ where+ cli = case ver of+ 3 -> v3 su V3+ 2 -> v2 su V2+ 1 -> v1 su V1+ 0 -> v0 su V0+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|>+ _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|>+ v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI+#else+clientLockContent _ = ()+#endif++clientKeepLocked+ :: LockID+ -> UUID+ -> a+ -> (VerifiedCopy -> Annex a)+ -- ^ Callback is run only after successfully connecting to the http+ -- server. The lock will remain held until the callback returns,+ -- and then will be dropped.+ -> ClientAction a+#ifdef WITH_SERVANT+clientKeepLocked lckid remoteuuid unablelock callback clientenv (ProtocolVersion ver) su cu bypass auth = do+ readyv <- liftIO newEmptyTMVarIO+ keeplocked <- liftIO newEmptyTMVarIO+ let cli' = cli lckid (Just cu) bypass auth+ (Just connectionKeepAlive) (Just keepAlive)+ (S.fromStepT (unlocksender readyv keeplocked))+ starttime <- liftIO getPOSIXTime+ tid <- liftIO $ async $ withClientM cli' clientenv $ \case+ Right (LockResult _ _) -> + atomically $ writeTMVar readyv (Right False)+ Left err -> + atomically $ writeTMVar readyv (Left err)+ let releaselock = liftIO $ do+ atomically $ putTMVar keeplocked False+ wait tid+ liftIO (atomically $ takeTMVar readyv) >>= \case+ Left err -> do+ liftIO $ wait tid+ return (Left err)+ Right False -> do+ liftIO $ wait tid+ return (Right unablelock)+ Right True -> do+ let checker = return $ Left $ starttime + retentionduration+ Right + <$> withVerifiedCopy LockedCopy remoteuuid checker callback+ `finally` releaselock+ where+ retentionduration = fromIntegral $+ durationSeconds p2pDefaultLockContentRetentionDuration++ unlocksender readyv keeplocked =+ S.Yield (UnlockRequest False) $ S.Effect $ do+ return $ S.Effect $ do+ liftIO $ atomically $ void $+ tryPutTMVar readyv (Right True)+ stilllocked <- liftIO $ atomically $+ takeTMVar keeplocked+ return $ if stilllocked+ then unlocksender readyv keeplocked+ else S.Yield (UnlockRequest True) S.Stop+ + cli = case ver of+ 3 -> v3 su V3+ 2 -> v2 su V2+ 1 -> v1 su V1+ 0 -> v0 su V0+ _ -> error "unsupported protocol version"+ + _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|>+ _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|>+ _ :<|> _ :<|> _ :<|> _ :<|>+ v3 :<|> v2 :<|> v1 :<|> v0 :<|> _ = client p2pHttpAPI+#else+clientKeepLocked _ _ _ _ = ()+#endif
+ P2P/Http/Server.hs view
@@ -0,0 +1,478 @@+{- P2P protocol over HTTP, server+ -+ - https://git-annex.branchable.com/design/p2p_protocol_over_http/+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE DataKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-}++module P2P.Http.Server (+ module P2P.Http,+ module P2P.Http.Server,+ module P2P.Http.Types,+ module P2P.Http.State,+) where++import Annex.Common+import P2P.Http+import P2P.Http.Types+import P2P.Http.State+import P2P.Protocol hiding (Offset, Bypass, auth)+import P2P.IO+import P2P.Annex+import Annex.WorkerPool+import Types.WorkerPool+import Types.Direction+import Utility.Metered++import Servant+import qualified Servant.Types.SourceT as S+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as L+import qualified Data.ByteString.Lazy.Internal as LI+import Control.Concurrent.STM+import Control.Concurrent.Async+import Control.Concurrent+import System.IO.Unsafe+import Data.Either++p2pHttpApp :: P2PHttpServerState -> Application+p2pHttpApp = serve p2pHttpAPI . serveP2pHttp++serveP2pHttp :: P2PHttpServerState -> Server P2PHttpAPI+serveP2pHttp st+ = serveGet st+ :<|> serveGet st+ :<|> serveGet st+ :<|> serveGet st+ :<|> serveCheckPresent st+ :<|> serveCheckPresent st+ :<|> serveCheckPresent st+ :<|> serveCheckPresent st+ :<|> serveRemove st id+ :<|> serveRemove st id+ :<|> serveRemove st dePlus+ :<|> serveRemove st dePlus+ :<|> serveRemoveBefore st+ :<|> serveGetTimestamp st+ :<|> servePut st id+ :<|> servePut st id+ :<|> servePut st dePlus+ :<|> servePut st dePlus+ :<|> servePutOffset st id+ :<|> servePutOffset st id+ :<|> servePutOffset st dePlus+ :<|> serveLockContent st+ :<|> serveLockContent st+ :<|> serveLockContent st+ :<|> serveLockContent st+ :<|> serveKeepLocked st+ :<|> serveKeepLocked st+ :<|> serveKeepLocked st+ :<|> serveKeepLocked st+ :<|> serveGetGeneric st++serveGetGeneric+ :: P2PHttpServerState+ -> B64UUID ServerSide+ -> B64Key+ -> Maybe (B64UUID ClientSide)+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString))+serveGetGeneric st su@(B64UUID u) k mcu bypass =+ -- Use V0 because it does not alter the returned data to indicate+ -- Invalid content.+ serveGet st su V0 k (fromMaybe scu mcu) bypass Nothing Nothing+ where+ -- Reuse server UUID as client UUID.+ scu = B64UUID u :: B64UUID ClientSide++serveGet+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> Maybe B64FilePath+ -> Maybe Offset+ -> IsSecure+ -> Maybe Auth+ -> Handler (Headers '[DataLengthHeader] (S.SourceT IO B.ByteString))+serveGet st su apiver (B64Key k) cu bypass baf startat sec auth = do+ conn <- getP2PConnection apiver st cu su bypass sec auth ReadAction id+ bsv <- liftIO newEmptyTMVarIO+ endv <- liftIO newEmptyTMVarIO+ validityv <- liftIO newEmptyTMVarIO+ finalv <- liftIO newEmptyTMVarIO+ annexworker <- liftIO $ async $ inAnnexWorker st $ do+ let storer _offset len = sendContentWith $ \bs -> liftIO $ do+ atomically $ putTMVar bsv (len, bs)+ atomically $ takeTMVar endv+ signalFullyConsumedByteString $+ connOhdl $ serverP2PConnection conn+ return $ \v -> do+ liftIO $ atomically $ putTMVar validityv v+ return True+ enteringStage (TransferStage Upload) $+ runFullProto (clientRunState conn) (clientP2PConnection conn) $+ void $ receiveContent Nothing nullMeterUpdate+ sizer storer getreq+ void $ liftIO $ forkIO $ waitfinal endv finalv conn annexworker+ (Len len, bs) <- liftIO $ atomically $ takeTMVar bsv+ bv <- liftIO $ newMVar (filter (not . B.null) (L.toChunks bs))+ szv <- liftIO $ newMVar 0+ let streamer = S.SourceT $ \s -> s =<< return + (stream (bv, szv, len, endv, validityv, finalv))+ return $ addHeader (DataLength len) streamer+ where+ stream (bv, szv, len, endv, validityv, finalv) =+ S.fromActionStep B.null $+ modifyMVar bv $ nextchunk szv $+ checkvalidity szv len endv validityv finalv+ + nextchunk szv checkvalid (b:[]) = do+ updateszv szv b+ ifM checkvalid+ ( return ([], b)+ -- The key's content is invalid, but+ -- the amount of data is the same as the+ -- DataLengthHeader indicated. Truncate+ -- the response by one byte to indicate+ -- to the client that it's not valid.+ , return ([], B.take (B.length b - 1) b)+ )+ nextchunk szv _checkvalid (b:bs) = do+ updateszv szv b+ return (bs, b)+ nextchunk _szv checkvalid [] = do+ void checkvalid+ -- Result ignored because 0 bytes of data are sent,+ -- so even if the key is invalid, if that's the+ -- amount of data that the DataLengthHeader indicates,+ -- we've successfully served an empty key.+ return ([], mempty)+ + updateszv szv b = modifyMVar szv $ \sz ->+ let !sz' = sz + fromIntegral (B.length b)+ in return (sz', ())++ -- Returns False when the key's content is invalid, but the+ -- amount of data sent was the same as indicated by the+ -- DataLengthHeader.+ checkvalidity szv len endv validityv finalv =+ ifM (atomically $ isEmptyTMVar endv)+ ( do+ atomically $ putTMVar endv ()+ validity <- atomically $ takeTMVar validityv+ sz <- takeMVar szv+ atomically $ putTMVar finalv ()+ atomically $ putTMVar endv ()+ return $ case validity of+ Nothing -> True+ Just Valid -> True+ Just Invalid -> sz /= len+ , pure True+ )+ + waitfinal endv finalv conn annexworker = do+ -- Wait for everything to be transferred before+ -- stopping the annexworker. The finalv will usually+ -- be written to at the end. If the client disconnects+ -- early that does not happen, so catch STM exception.+ alltransferred <- isRight+ <$> tryNonAsync (liftIO $ atomically $ takeTMVar finalv)+ -- Make sure the annexworker is not left blocked on endv+ -- if the client disconnected early.+ void $ liftIO $ atomically $ tryPutTMVar endv ()+ void $ tryNonAsync $ if alltransferred+ then releaseP2PConnection conn+ else closeP2PConnection conn+ void $ tryNonAsync $ wait annexworker+ + sizer = pure $ Len $ case startat of+ Just (Offset o) -> fromIntegral o+ Nothing -> 0+ + getreq offset = P2P.Protocol.GET offset af k+ + af = ProtoAssociatedFile $ b64FilePathToAssociatedFile baf++serveCheckPresent+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler CheckPresentResult+serveCheckPresent st su apiver (B64Key k) cu bypass sec auth = do+ res <- withP2PConnection apiver st cu su bypass sec auth ReadAction id+ $ \conn -> liftIO $ proxyClientNetProto conn $ checkPresent k+ case res of+ Right b -> return (CheckPresentResult b)+ Left err -> throwError $ err500 { errBody = encodeBL err }++serveRemove+ :: APIVersion v+ => P2PHttpServerState+ -> (RemoveResultPlus -> t)+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler t+serveRemove st resultmangle su apiver (B64Key k) cu bypass sec auth = do+ res <- withP2PConnection apiver st cu su bypass sec auth RemoveAction id+ $ \conn ->+ liftIO $ proxyClientNetProto conn $ remove Nothing k+ case res of+ (Right b, plusuuids) -> return $ resultmangle $ + RemoveResultPlus b (map B64UUID (fromMaybe [] plusuuids))+ (Left err, _) -> throwError $+ err500 { errBody = encodeBL err }++serveRemoveBefore+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> Timestamp+ -> IsSecure+ -> Maybe Auth+ -> Handler RemoveResultPlus+serveRemoveBefore st su apiver (B64Key k) cu bypass (Timestamp ts) sec auth = do+ res <- withP2PConnection apiver st cu su bypass sec auth RemoveAction id+ $ \conn ->+ liftIO $ proxyClientNetProto conn $+ removeBeforeRemoteEndTime ts k+ case res of+ (Right b, plusuuids) -> return $ + RemoveResultPlus b (map B64UUID (fromMaybe [] plusuuids))+ (Left err, _) -> throwError $+ err500 { errBody = encodeBL err }++serveGetTimestamp+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler GetTimestampResult+serveGetTimestamp st su apiver cu bypass sec auth = do+ res <- withP2PConnection apiver st cu su bypass sec auth ReadAction id+ $ \conn ->+ liftIO $ proxyClientNetProto conn getTimestamp+ case res of+ Right ts -> return $ GetTimestampResult (Timestamp ts)+ Left err -> throwError $+ err500 { errBody = encodeBL err }++servePut+ :: APIVersion v+ => P2PHttpServerState+ -> (PutResultPlus -> t)+ -> B64UUID ServerSide+ -> v+ -> DataLength+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> Maybe B64FilePath+ -> Maybe Offset+ -> S.SourceT IO B.ByteString+ -> IsSecure+ -> Maybe Auth+ -> Handler t+servePut st resultmangle su apiver (DataLength len) (B64Key k) cu bypass baf moffset stream sec auth = do+ validityv <- liftIO newEmptyTMVarIO+ let validitycheck = local $ runValidityCheck $+ liftIO $ atomically $ readTMVar validityv+ tooshortv <- liftIO newEmptyTMVarIO+ content <- liftIO $ S.unSourceT stream (gather validityv tooshortv)+ res <- withP2PConnection' apiver st cu su bypass sec auth WriteAction+ (\cst -> cst { connectionWaitVar = False }) $ \conn -> do+ liftIO $ void $ async $ checktooshort conn tooshortv+ liftIO (protoaction conn content validitycheck)+ case res of+ Right (Right (Just plusuuids)) -> return $ resultmangle $+ PutResultPlus True (map B64UUID plusuuids)+ Right (Right Nothing) -> return $ resultmangle $+ PutResultPlus False []+ Right (Left protofail) -> throwError $+ err500 { errBody = encodeBL (describeProtoFailure protofail) }+ Left err -> throwError $+ err500 { errBody = encodeBL (show err) }+ where+ protoaction conn content validitycheck = inAnnexWorker st $+ enteringStage (TransferStage Download) $+ runFullProto (clientRunState conn) (clientP2PConnection conn) $+ protoaction' content validitycheck+ + protoaction' content validitycheck = put' k af $ \offset' ->+ let offsetdelta = offset' - offset+ in case compare offset' offset of+ EQ -> sendContent' nullMeterUpdate (Len len)+ content validitycheck+ GT -> sendContent' nullMeterUpdate+ (Len (len - fromIntegral offsetdelta))+ (L.drop (fromIntegral offsetdelta) content)+ validitycheck+ LT -> sendContent' nullMeterUpdate+ (Len len)+ content+ (validitycheck >>= \_ -> return Invalid)+ + offset = case moffset of+ Just (Offset o) -> o+ Nothing -> 0++ af = b64FilePathToAssociatedFile baf++ -- Streams the ByteString from the client. Avoids returning a longer+ -- than expected ByteString by truncating to the expected length. + -- Returns a shorter than expected ByteString when the data is not+ -- valid.+ gather validityv tooshortv = unsafeInterleaveIO . go 0+ where+ go n S.Stop = do+ atomically $ do+ writeTMVar validityv $+ if n == len then Valid else Invalid+ writeTMVar tooshortv (n /= len)+ return LI.Empty+ go n (S.Error _err) = do+ atomically $ do+ writeTMVar validityv Invalid+ writeTMVar tooshortv (n /= len)+ return LI.Empty+ go n (S.Skip s) = go n s+ go n (S.Effect ms) = ms >>= go n+ go n (S.Yield v s) =+ let !n' = n + fromIntegral (B.length v)+ in if n' > len+ then do+ atomically $ do+ writeTMVar validityv Invalid+ writeTMVar tooshortv True+ return $ LI.Chunk+ (B.take (fromIntegral (len - n')) v)+ LI.Empty+ else LI.Chunk v <$> unsafeInterleaveIO (go n' s)+ + -- The connection can no longer be used when too short a DATA has+ -- been written to it.+ checktooshort conn tooshortv = do+ liftIO $ whenM (atomically $ takeTMVar tooshortv) $+ closeP2PConnection conn++servePutOffset+ :: APIVersion v+ => P2PHttpServerState+ -> (PutOffsetResultPlus -> t)+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler t+servePutOffset st resultmangle su apiver (B64Key k) cu bypass sec auth = do+ res <- withP2PConnection apiver st cu su bypass sec auth WriteAction+ (\cst -> cst { connectionWaitVar = False }) $ \conn ->+ liftIO $ proxyClientNetProto conn $ getPutOffset k af+ case res of+ Right offset -> return $ resultmangle $+ PutOffsetResultPlus (Offset offset)+ Left plusuuids -> return $ resultmangle $+ PutOffsetResultAlreadyHavePlus (map B64UUID plusuuids)+ where+ af = AssociatedFile Nothing++serveLockContent+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> B64Key+ -> B64UUID ClientSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Handler LockResult+serveLockContent st su apiver (B64Key k) cu bypass sec auth = do+ conn <- getP2PConnection apiver st cu su bypass sec auth WriteAction id+ let lock = do+ lockresv <- newEmptyTMVarIO+ unlockv <- newEmptyTMVarIO+ annexworker <- async $ inAnnexWorker st $ do+ lockres <- runFullProto (clientRunState conn) (clientP2PConnection conn) $ do+ net $ sendMessage (LOCKCONTENT k)+ checkSuccess+ liftIO $ atomically $ putTMVar lockresv lockres+ liftIO $ atomically $ takeTMVar unlockv+ void $ runFullProto (clientRunState conn) (clientP2PConnection conn) $ do+ net $ sendMessage UNLOCKCONTENT+ atomically (takeTMVar lockresv) >>= \case+ Right True -> return (Just (annexworker, unlockv))+ _ -> return Nothing+ let unlock (annexworker, unlockv) = do+ atomically $ putTMVar unlockv ()+ void $ wait annexworker+ releaseP2PConnection conn+ liftIO $ mkLocker lock unlock >>= \case+ Just (locker, lockid) -> do+ liftIO $ storeLock lockid locker st+ return $ LockResult True (Just lockid)+ Nothing -> return $ LockResult False Nothing++serveKeepLocked+ :: APIVersion v+ => P2PHttpServerState+ -> B64UUID ServerSide+ -> v+ -> LockID+ -> Maybe (B64UUID ClientSide)+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> Maybe ConnectionKeepAlive+ -> Maybe KeepAlive+ -> S.SourceT IO UnlockRequest+ -> Handler LockResult+serveKeepLocked st _su _apiver lckid _cu _bypass sec auth _ _ unlockrequeststream = do+ checkAuthActionClass st sec auth WriteAction $ \_ -> do+ liftIO $ keepingLocked lckid st+ _ <- liftIO $ S.unSourceT unlockrequeststream go+ return (LockResult False Nothing)+ where+ go S.Stop = dropLock lckid st+ go (S.Error _err) = dropLock lckid st+ go (S.Skip s) = go s+ go (S.Effect ms) = ms >>= go+ go (S.Yield (UnlockRequest False) s) = go s+ go (S.Yield (UnlockRequest True) _) = dropLock lckid st
+ P2P/Http/State.hs view
@@ -0,0 +1,657 @@+{- P2P protocol over HTTP, server state+ -+ - https://git-annex.branchable.com/design/p2p_protocol_over_http/+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE TupleSections #-}++module P2P.Http.State where++import Annex.Common+import qualified Annex+import P2P.Http.Types+import qualified P2P.Protocol as P2P+import qualified P2P.IO as P2P+import P2P.IO+import P2P.Annex+import Annex.UUID+import Types.NumCopies+import Types.WorkerPool+import Annex.WorkerPool+import Annex.BranchState+import Types.Cluster+import CmdLine.Action (startConcurrency)+import Utility.ThreadScheduler+import Utility.HumanTime+import Logs.Proxy+import Annex.Proxy+import Annex.Cluster+import qualified P2P.Proxy as Proxy+import qualified Types.Remote as Remote++import Servant+import qualified Data.Map.Strict as M+import qualified Data.Set as S+import Control.Concurrent.Async+import Control.Concurrent.STM+import Data.Time.Clock.POSIX++data P2PHttpServerState = P2PHttpServerState+ { acquireP2PConnection :: AcquireP2PConnection+ , annexWorkerPool :: AnnexWorkerPool+ , getServerMode :: GetServerMode+ , openLocks :: TMVar (M.Map LockID Locker)+ }++type AnnexWorkerPool = TMVar (WorkerPool (Annex.AnnexState, Annex.AnnexRead))++-- Nothing when the server is not allowed to serve any requests.+type GetServerMode = IsSecure -> Maybe Auth -> Maybe P2P.ServerMode++mkP2PHttpServerState :: AcquireP2PConnection -> AnnexWorkerPool -> GetServerMode -> IO P2PHttpServerState+mkP2PHttpServerState acquireconn annexworkerpool getservermode = P2PHttpServerState+ <$> pure acquireconn+ <*> pure annexworkerpool+ <*> pure getservermode+ <*> newTMVarIO mempty++data ActionClass = ReadAction | WriteAction | RemoveAction+ deriving (Eq)++withP2PConnection+ :: APIVersion v+ => v+ -> P2PHttpServerState+ -> B64UUID ClientSide+ -> B64UUID ServerSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> ActionClass+ -> (ConnectionParams -> ConnectionParams)+ -> (P2PConnectionPair -> Handler (Either ProtoFailure a))+ -> Handler a+withP2PConnection apiver st cu su bypass sec auth actionclass fconnparams connaction =+ withP2PConnection' apiver st cu su bypass sec auth actionclass fconnparams connaction'+ where+ connaction' conn = connaction conn >>= \case+ Right r -> return r+ Left err -> throwError $+ err500 { errBody = encodeBL (describeProtoFailure err) }++withP2PConnection'+ :: APIVersion v+ => v+ -> P2PHttpServerState+ -> B64UUID ClientSide+ -> B64UUID ServerSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> ActionClass+ -> (ConnectionParams -> ConnectionParams)+ -> (P2PConnectionPair -> Handler a)+ -> Handler a+withP2PConnection' apiver st cu su bypass sec auth actionclass fconnparams connaction = do+ conn <- getP2PConnection apiver st cu su bypass sec auth actionclass fconnparams+ connaction conn+ `finally` liftIO (releaseP2PConnection conn)++getP2PConnection+ :: APIVersion v+ => v+ -> P2PHttpServerState+ -> B64UUID ClientSide+ -> B64UUID ServerSide+ -> [B64UUID Bypass]+ -> IsSecure+ -> Maybe Auth+ -> ActionClass+ -> (ConnectionParams -> ConnectionParams)+ -> Handler P2PConnectionPair+getP2PConnection apiver st cu su bypass sec auth actionclass fconnparams =+ checkAuthActionClass st sec auth actionclass go+ where+ go servermode = liftIO (acquireP2PConnection st cp) >>= \case+ Left (ConnectionFailed err) -> + throwError err502 { errBody = encodeBL err }+ Left TooManyConnections ->+ throwError err503+ Right v -> return v+ where+ cp = fconnparams $ ConnectionParams+ { connectionProtocolVersion = protocolVersion apiver+ , connectionServerUUID = fromB64UUID su+ , connectionClientUUID = fromB64UUID cu+ , connectionBypass = map fromB64UUID bypass+ , connectionServerMode = servermode+ , connectionWaitVar = True+ }++checkAuthActionClass+ :: P2PHttpServerState+ -> IsSecure+ -> Maybe Auth+ -> ActionClass+ -> (P2P.ServerMode -> Handler a)+ -> Handler a+checkAuthActionClass st sec auth actionclass go =+ case (getServerMode st sec auth, actionclass) of+ (Just P2P.ServeReadWrite, _) -> go P2P.ServeReadWrite+ (Just P2P.ServeAppendOnly, RemoveAction) -> throwError err403+ (Just P2P.ServeAppendOnly, _) -> go P2P.ServeAppendOnly+ (Just P2P.ServeReadOnly, ReadAction) -> go P2P.ServeReadOnly+ (Just P2P.ServeReadOnly, _) -> throwError err403+ (Nothing, _) -> throwError basicAuthRequired++basicAuthRequired :: ServerError+basicAuthRequired = err401 { errHeaders = [(h, v)] }+ where+ h = "WWW-Authenticate"+ v = "Basic realm=\"git-annex\", charset=\"UTF-8\""++data ConnectionParams = ConnectionParams+ { connectionProtocolVersion :: P2P.ProtocolVersion+ , connectionServerUUID :: UUID+ , connectionClientUUID :: UUID+ , connectionBypass :: [UUID]+ , connectionServerMode :: P2P.ServerMode+ , connectionWaitVar :: Bool+ }+ deriving (Show, Eq, Ord)++data ConnectionProblem+ = ConnectionFailed String+ | TooManyConnections+ deriving (Show, Eq)++proxyClientNetProto :: P2PConnectionPair -> P2P.Proto a -> IO (Either P2P.ProtoFailure a)+proxyClientNetProto conn = runNetProto+ (clientRunState conn) (clientP2PConnection conn)++type AcquireP2PConnection+ = ConnectionParams+ -> IO (Either ConnectionProblem P2PConnectionPair)++withP2PConnections+ :: AnnexWorkerPool+ -> ProxyConnectionPoolSize+ -> ClusterConcurrency+ -> (AcquireP2PConnection -> Annex a)+ -> Annex a+withP2PConnections workerpool proxyconnectionpoolsize clusterconcurrency a = do+ enableInteractiveBranchAccess+ myuuid <- getUUID+ myproxies <- M.lookup myuuid <$> getProxies+ reqv <- liftIO newEmptyTMVarIO+ relv <- liftIO newEmptyTMVarIO+ endv <- liftIO newEmptyTMVarIO+ proxypool <- liftIO $ newTMVarIO (0, mempty)+ asyncservicer <- liftIO $ async $+ servicer myuuid myproxies proxypool reqv relv endv+ let endit = do+ liftIO $ atomically $ putTMVar endv ()+ liftIO $ wait asyncservicer+ a (acquireconn reqv) `finally` endit+ where+ acquireconn reqv connparams = do+ respvar <- newEmptyTMVarIO+ atomically $ putTMVar reqv (connparams, respvar)+ atomically $ takeTMVar respvar++ servicer myuuid myproxies proxypool reqv relv endv = do+ reqrel <- liftIO $+ atomically $ + (Right <$> takeTMVar reqv)+ `orElse` + (Left . Right <$> takeTMVar relv)+ `orElse` + (Left . Left <$> takeTMVar endv)+ case reqrel of+ Right (connparams, respvar) -> do+ servicereq myuuid myproxies proxypool relv connparams+ >>= atomically . putTMVar respvar+ servicer myuuid myproxies proxypool reqv relv endv+ Left (Right releaseconn) -> do+ void $ tryNonAsync releaseconn+ servicer myuuid myproxies proxypool reqv relv endv+ Left (Left ()) -> return ()+ + servicereq myuuid myproxies proxypool relv connparams+ | connectionServerUUID connparams == myuuid =+ localConnection relv connparams workerpool+ | otherwise =+ atomically (getProxyConnectionPool proxypool connparams) >>= \case+ Just conn -> proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool conn+ Nothing -> checkcanproxy myproxies proxypool relv connparams++ checkcanproxy myproxies proxypool relv connparams = + inAnnexWorker' workerpool+ (checkCanProxy' myproxies (connectionServerUUID connparams))+ >>= \case+ Right (Left reason) -> return $ Left $+ ConnectionFailed $ + fromMaybe "unknown uuid" reason+ Right (Right (Right proxyremote)) -> proxyconnection $+ openProxyConnectionToRemote workerpool+ (connectionProtocolVersion connparams)+ bypass proxyremote+ Right (Right (Left clusteruuid)) -> proxyconnection $+ openProxyConnectionToCluster workerpool+ (connectionProtocolVersion connparams)+ bypass clusteruuid clusterconcurrency+ Left ex -> return $ Left $+ ConnectionFailed $ show ex+ where+ bypass = P2P.Bypass $ S.fromList $ connectionBypass connparams+ proxyconnection openconn = openconn >>= \case+ Right conn -> proxyConnection proxyconnectionpoolsize+ relv connparams workerpool proxypool conn+ Left ex -> return $ Left $+ ConnectionFailed $ show ex++data P2PConnectionPair = P2PConnectionPair+ { clientRunState :: RunState+ , clientP2PConnection :: P2PConnection+ , serverP2PConnection :: P2PConnection+ , releaseP2PConnection :: IO ()+ -- ^ Releases a P2P connection, which can be reused for other+ -- requests.+ , closeP2PConnection :: IO ()+ -- ^ Closes a P2P connection, which is in a state where it is no+ -- longer usable.+ }++localConnection+ :: TMVar (IO ())+ -> ConnectionParams+ -> AnnexWorkerPool+ -> IO (Either ConnectionProblem P2PConnectionPair)+localConnection relv connparams workerpool = + localP2PConnectionPair connparams relv $ \serverrunst serverconn ->+ inAnnexWorker' workerpool $+ void $ runFullProto serverrunst serverconn $+ P2P.serveOneCommandAuthed+ (connectionServerMode connparams)+ (connectionServerUUID connparams)++localP2PConnectionPair+ :: ConnectionParams+ -> TMVar (IO ())+ -> (RunState -> P2PConnection -> IO (Either SomeException ()))+ -> IO (Either ConnectionProblem P2PConnectionPair)+localP2PConnectionPair connparams relv startworker = do+ (clientconn, serverconn) <- mkP2PConnectionPair connparams+ ("http client", "http server")+ clientrunst <- mkClientRunState connparams+ serverrunst <- mkServerRunState connparams+ asyncworker <- async $+ startworker serverrunst serverconn+ let releaseconn = atomically $ void $ tryPutTMVar relv $+ liftIO $ wait asyncworker+ >>= either throwM return+ return $ Right $ P2PConnectionPair+ { clientRunState = clientrunst+ , clientP2PConnection = clientconn+ , serverP2PConnection = serverconn+ , releaseP2PConnection = releaseconn+ , closeP2PConnection = releaseconn+ }++mkP2PConnectionPair+ :: ConnectionParams+ -> (String, String)+ -> IO (P2PConnection, P2PConnection)+mkP2PConnectionPair connparams (n1, n2) = do+ hdl1 <- newEmptyTMVarIO+ hdl2 <- newEmptyTMVarIO+ wait1 <- newEmptyTMVarIO+ wait2 <- newEmptyTMVarIO+ closed1 <- newEmptyTMVarIO+ closed2 <- newEmptyTMVarIO+ let h1 = P2PHandleTMVar hdl1+ (if connectionWaitVar connparams then Just wait1 else Nothing)+ closed1+ let h2 = P2PHandleTMVar hdl2+ (if connectionWaitVar connparams then Just wait2 else Nothing)+ closed2+ let clientconn = P2PConnection Nothing+ (const True) h2 h1+ (ConnIdent (Just n1))+ let serverconn = P2PConnection Nothing+ (const True) h1 h2+ (ConnIdent (Just n2))+ return (clientconn, serverconn)++mkServerRunState :: ConnectionParams -> IO RunState+mkServerRunState connparams = do+ prototvar <- newTVarIO $ connectionProtocolVersion connparams+ mkRunState $ const $ Serving + (connectionClientUUID connparams)+ Nothing+ prototvar+ +mkClientRunState :: ConnectionParams -> IO RunState+mkClientRunState connparams = do+ prototvar <- newTVarIO $ connectionProtocolVersion connparams+ mkRunState $ const $ Client prototvar++proxyConnection+ :: ProxyConnectionPoolSize+ -> TMVar (IO ())+ -> ConnectionParams+ -> AnnexWorkerPool+ -> TMVar ProxyConnectionPool+ -> ProxyConnection+ -> IO (Either ConnectionProblem P2PConnectionPair)+proxyConnection proxyconnectionpoolsize relv connparams workerpool proxypool proxyconn = do+ (clientconn, proxyfromclientconn) <- + mkP2PConnectionPair connparams ("http client", "proxy")+ clientrunst <- mkClientRunState connparams+ proxyfromclientrunst <- mkClientRunState connparams+ asyncworker <- async $+ inAnnexWorker' workerpool $ do+ proxystate <- liftIO Proxy.mkProxyState+ let proxyparams = Proxy.ProxyParams+ { Proxy.proxyMethods = mkProxyMethods+ , Proxy.proxyState = proxystate+ , Proxy.proxyServerMode = connectionServerMode connparams+ , Proxy.proxyClientSide = Proxy.ClientSide proxyfromclientrunst proxyfromclientconn+ , Proxy.proxyUUID = proxyConnectionRemoteUUID proxyconn+ , Proxy.proxySelector = proxyConnectionSelector proxyconn+ , Proxy.proxyConcurrencyConfig = proxyConnectionConcurrency proxyconn+ , Proxy.proxyClientProtocolVersion = connectionProtocolVersion connparams+ }+ let proxy mrequestmessage = case mrequestmessage of+ Just requestmessage -> do+ Proxy.proxyRequest proxydone proxyparams+ requestcomplete requestmessage protoerrhandler+ Nothing -> return ()+ protoerrhandler proxy $+ liftIO $ runNetProto proxyfromclientrunst proxyfromclientconn $+ P2P.net P2P.receiveMessage+ + let closebothsides = do+ liftIO $ closeConnection proxyfromclientconn+ liftIO $ closeConnection clientconn++ let releaseconn connstillusable = do+ atomically $ void $ tryPutTMVar relv $ do+ unless connstillusable+ closebothsides+ r <- liftIO $ wait asyncworker+ when connstillusable+ closebothsides+ if connstillusable+ then liftIO $ do+ now <- getPOSIXTime+ evicted <- atomically $ putProxyConnectionPool proxypool proxyconnectionpoolsize connparams $+ proxyconn { proxyConnectionLastUsed = now }+ maybe noop closeproxyconnection evicted+ else closeproxyconnection proxyconn+ either throwM return r+ + return $ Right $ P2PConnectionPair+ { clientRunState = clientrunst+ , clientP2PConnection = clientconn+ , serverP2PConnection = proxyfromclientconn+ , releaseP2PConnection = releaseconn True+ , closeP2PConnection = releaseconn False+ }+ where+ protoerrhandler cont a = a >>= \case+ Left _ -> proxyConnectionCloser proxyconn+ Right v -> cont v+ + proxydone = return ()+ + requestcomplete () = return ()+ + closeproxyconnection = + void . inAnnexWorker' workerpool . proxyConnectionCloser++data Locker = Locker+ { lockerThread :: Async ()+ , lockerVar :: TMVar Bool+ -- ^ Left empty until the thread has taken the lock+ -- (or failed to do so), then True while the lock is held,+ -- and setting to False causes the lock to be released.+ , lockerTimeoutDisable :: TMVar ()+ -- ^ Until this is filled, the lock will be subject to timeout.+ -- Once filled the lock will remain held until explicitly dropped.+ }++mkLocker :: (IO (Maybe a)) -> (a -> IO ()) -> IO (Maybe (Locker, LockID))+mkLocker lock unlock = do+ lv <- newEmptyTMVarIO+ timeoutdisablev <- newEmptyTMVarIO+ let setlocked = putTMVar lv+ locktid <- async $ lock >>= \case+ Nothing ->+ atomically $ setlocked False+ Just st -> do+ atomically $ setlocked True+ atomically $ do+ v <- takeTMVar lv+ if v+ then retry+ else setlocked False+ unlock st+ locksuccess <- atomically $ readTMVar lv+ if locksuccess+ then do+ timeouttid <- async $ do+ threadDelaySeconds $ Seconds $ fromIntegral $+ durationSeconds p2pDefaultLockContentRetentionDuration+ atomically (tryReadTMVar timeoutdisablev) >>= \case+ Nothing -> void $ atomically $+ writeTMVar lv False+ Just () -> noop+ tid <- async $ do+ wait locktid+ cancel timeouttid+ lckid <- B64UUID <$> genUUID+ return (Just (Locker tid lv timeoutdisablev, lckid))+ else do+ wait locktid+ return Nothing++storeLock :: LockID -> Locker -> P2PHttpServerState -> IO ()+storeLock lckid locker st = atomically $ do+ m <- takeTMVar (openLocks st)+ let !m' = M.insert lckid locker m+ putTMVar (openLocks st) m'++keepingLocked :: LockID -> P2PHttpServerState -> IO ()+keepingLocked lckid st = do+ m <- atomically $ readTMVar (openLocks st)+ case M.lookup lckid m of+ Nothing -> return ()+ Just locker ->+ atomically $ void $ + tryPutTMVar (lockerTimeoutDisable locker) ()++dropLock :: LockID -> P2PHttpServerState -> IO ()+dropLock lckid st = do+ v <- atomically $ do+ m <- takeTMVar (openLocks st)+ let (mlocker, !m') =+ M.updateLookupWithKey (\_ _ -> Nothing) lckid m+ putTMVar (openLocks st) m'+ case mlocker of+ Nothing -> return Nothing+ -- Signal to the locker's thread that it can+ -- release the lock.+ Just locker -> do+ _ <- swapTMVar (lockerVar locker) False+ return (Just locker)+ case v of+ Nothing -> return ()+ Just locker -> wait (lockerThread locker)++getAnnexWorkerPool :: (AnnexWorkerPool -> Annex a) -> Annex a+getAnnexWorkerPool a = startConcurrency transferStages $+ Annex.getState Annex.workers >>= \case+ Nothing -> giveup "Use -Jn or set annex.jobs to configure the number of worker threads."+ Just wp -> a wp++inAnnexWorker :: P2PHttpServerState -> Annex a -> IO (Either SomeException a)+inAnnexWorker st = inAnnexWorker' (annexWorkerPool st)++inAnnexWorker' :: AnnexWorkerPool -> Annex a -> IO (Either SomeException a)+inAnnexWorker' poolv annexaction = do+ (workerstrd, workerstage) <- atomically $ waitStartWorkerSlot poolv+ resv <- newEmptyTMVarIO+ aid <- async $ do+ (res, strd) <- Annex.run workerstrd annexaction+ atomically $ putTMVar resv res+ return strd+ atomically $ do+ pool <- takeTMVar poolv+ let !pool' = addWorkerPool (ActiveWorker aid workerstage) pool+ putTMVar poolv pool'+ (res, workerstrd') <- waitCatch aid >>= \case+ Right strd -> do+ r <- atomically $ takeTMVar resv+ return (Right r, strd)+ Left err -> return (Left err, workerstrd)+ atomically $ do+ pool <- takeTMVar poolv+ let !pool' = deactivateWorker pool aid workerstrd'+ putTMVar poolv pool'+ return res++data ProxyConnection = ProxyConnection+ { proxyConnectionRemoteUUID :: UUID+ , proxyConnectionSelector :: Proxy.ProxySelector+ , proxyConnectionCloser :: Annex ()+ , proxyConnectionConcurrency :: Proxy.ConcurrencyConfig+ , proxyConnectionLastUsed :: POSIXTime+ }++instance Show ProxyConnection where+ show pc = unwords+ [ "ProxyConnection"+ , show (proxyConnectionRemoteUUID pc)+ , show (proxyConnectionLastUsed pc)+ ]++openedProxyConnection+ :: UUID+ -> String+ -> Proxy.ProxySelector+ -> Annex ()+ -> Proxy.ConcurrencyConfig+ -> Annex ProxyConnection+openedProxyConnection u desc selector closer concurrency = do+ now <- liftIO getPOSIXTime+ fastDebug "P2P.Http" ("Opened proxy connection to " ++ desc)+ return $ ProxyConnection u selector closer' concurrency now+ where+ closer' = do+ fastDebug "P2P.Http" ("Closing proxy connection to " ++ desc)+ closer+ fastDebug "P2P.Http" ("Closed proxy connection to " ++ desc)++openProxyConnectionToRemote+ :: AnnexWorkerPool+ -> P2P.ProtocolVersion+ -> P2P.Bypass+ -> Remote+ -> IO (Either SomeException ProxyConnection)+openProxyConnectionToRemote workerpool clientmaxversion bypass remote =+ inAnnexWorker' workerpool $ do+ remoteside <- proxyRemoteSide clientmaxversion bypass remote+ concurrencyconfig <- Proxy.noConcurrencyConfig+ openedProxyConnection (Remote.uuid remote)+ ("remote " ++ Remote.name remote)+ (Proxy.singleProxySelector remoteside)+ (Proxy.closeRemoteSide remoteside)+ concurrencyconfig++type ClusterConcurrency = Int++openProxyConnectionToCluster+ :: AnnexWorkerPool+ -> P2P.ProtocolVersion+ -> P2P.Bypass+ -> ClusterUUID+ -> ClusterConcurrency+ -> IO (Either SomeException ProxyConnection)+openProxyConnectionToCluster workerpool clientmaxversion bypass clusteruuid concurrency =+ inAnnexWorker' workerpool $ do+ (proxyselector, closenodes) <-+ clusterProxySelector clusteruuid clientmaxversion bypass+ concurrencyconfig <- Proxy.mkConcurrencyConfig concurrency+ openedProxyConnection (fromClusterUUID clusteruuid)+ ("cluster " ++ fromUUID (fromClusterUUID clusteruuid))+ proxyselector closenodes concurrencyconfig++type ProxyConnectionPool = (Integer, M.Map ProxyConnectionPoolKey [ProxyConnection])++type ProxyConnectionPoolSize = Integer++-- Returns any older ProxyConnection that was evicted from the pool.+putProxyConnectionPool+ :: TMVar ProxyConnectionPool+ -> ProxyConnectionPoolSize+ -> ConnectionParams+ -> ProxyConnection+ -> STM (Maybe ProxyConnection)+putProxyConnectionPool proxypool maxsz connparams conn = do+ (sz, m) <- takeTMVar proxypool+ let ((sz', m'), evicted) = case M.lookup k m of+ Nothing -> ((succ sz, M.insert k [conn] m), Nothing)+ Just [] -> ((succ sz, M.insert k [conn] m), Nothing)+ Just cs -> if sz >= maxsz+ then ((sz, M.insert k (conn : dropFromEnd 1 cs) m), lastMaybe cs)+ else ((sz, M.insert k (conn : cs) m), Nothing)+ let ((sz'', m''), evicted') = if sz' > maxsz+ then removeOldestProxyConnectionPool (sz', m')+ else ((sz', m'), Nothing)+ putTMVar proxypool (sz'', m'')+ return (evicted <|> evicted')+ where+ k = proxyConnectionPoolKey connparams++removeOldestProxyConnectionPool :: ProxyConnectionPool -> (ProxyConnectionPool, Maybe ProxyConnection)+removeOldestProxyConnectionPool (sz, m) = + ((pred sz, m'), snd <$> headMaybe l)+ where+ m' = M.fromListWith (++) $ map (\(k', v) -> (k', [v])) (drop 1 l)+ l = sortOn (proxyConnectionLastUsed . snd) $+ concatMap (\(k', pl) -> map (k', ) pl) $+ M.toList m++getProxyConnectionPool+ :: TMVar ProxyConnectionPool+ -> ConnectionParams+ -> STM (Maybe ProxyConnection)+getProxyConnectionPool proxypool connparams = do+ (sz, m) <- takeTMVar proxypool+ case M.lookup k m of+ Just (c:cs) -> do+ putTMVar proxypool (sz-1, M.insert k cs m)+ return (Just c)+ _ -> do+ putTMVar proxypool (sz, m)+ return Nothing+ where+ k = proxyConnectionPoolKey connparams++type ProxyConnectionPoolKey = (UUID, UUID, [UUID], P2P.ProtocolVersion)++proxyConnectionPoolKey :: ConnectionParams -> ProxyConnectionPoolKey+proxyConnectionPoolKey connparams =+ ( connectionServerUUID connparams+ , connectionClientUUID connparams+ , connectionBypass connparams+ , connectionProtocolVersion connparams+ )
+ P2P/Http/Types.hs view
@@ -0,0 +1,402 @@+{- P2P protocol over HTTP,+ - data types for servant not including the servant API+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE CPP #-}++module P2P.Http.Types where++import Annex.Common+import qualified P2P.Protocol as P2P+import Utility.MonotonicClock++#ifdef WITH_SERVANT+import Servant+import Data.Aeson hiding (Key)+import Text.Read (readMaybe)+#endif+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import qualified Data.ByteString as B+import Codec.Binary.Base64Url as B64+import Data.Char+import Control.DeepSeq+import GHC.Generics (Generic)++data V3 = V3 deriving (Show)+data V2 = V2 deriving (Show)+data V1 = V1 deriving (Show)+data V0 = V0 deriving (Show)++class APIVersion v where+ protocolVersion :: v -> P2P.ProtocolVersion++instance APIVersion V3 where protocolVersion _ = P2P.ProtocolVersion 3+instance APIVersion V2 where protocolVersion _ = P2P.ProtocolVersion 2+instance APIVersion V1 where protocolVersion _ = P2P.ProtocolVersion 1+instance APIVersion V0 where protocolVersion _ = P2P.ProtocolVersion 0++-- Keys, UUIDs, and filenames can be base64 encoded since Servant uses +-- Text and so needs UTF-8.+newtype B64Key = B64Key Key+ deriving (Show)++newtype B64FilePath = B64FilePath RawFilePath+ deriving (Show)++associatedFileToB64FilePath :: AssociatedFile -> Maybe B64FilePath+associatedFileToB64FilePath (AssociatedFile Nothing) = Nothing+associatedFileToB64FilePath (AssociatedFile (Just f)) = Just (B64FilePath f)++b64FilePathToAssociatedFile :: Maybe B64FilePath -> AssociatedFile+b64FilePathToAssociatedFile Nothing = AssociatedFile Nothing+b64FilePathToAssociatedFile (Just (B64FilePath f)) = AssociatedFile (Just f)++newtype B64UUID t = B64UUID { fromB64UUID :: UUID }+ deriving (Show, Ord, Eq, Generic, NFData)++encodeB64Text :: B.ByteString -> T.Text+encodeB64Text b = case TE.decodeUtf8' b of+ Right t+ | (snd <$> B.unsnoc b) == Just closebracket + && (fst <$> B.uncons b) == Just openbracket ->+ b64wrapped+ | otherwise -> t+ Left _ -> b64wrapped+ where+ b64wrapped = TE.decodeUtf8 $ "[" <> B64.encode b <> "]"+ openbracket = fromIntegral (ord '[')+ closebracket = fromIntegral (ord ']')++decodeB64Text :: T.Text -> Either T.Text B.ByteString+decodeB64Text t = + case T.unsnoc t of+ Just (t', lastc) | lastc == ']' ->+ case T.uncons t' of+ Just (firstc, t'') | firstc == '[' ->+ case B64.decode (TE.encodeUtf8 t'') of+ Right b -> Right b+ Left _ -> Left "unable to base64 decode [] wrapped value"+ _ -> Right (TE.encodeUtf8 t)+ _ -> Right (TE.encodeUtf8 t)++-- Phantom types.+data ClientSide+data ServerSide+data Bypass+data Plus+data Lock++type LockID = B64UUID Lock++newtype DataLength = DataLength Integer+ deriving (Show)++newtype CheckPresentResult = CheckPresentResult Bool+ deriving (Show)++newtype RemoveResult = RemoveResult Bool+ deriving (Show)++data RemoveResultPlus = RemoveResultPlus Bool [B64UUID Plus]+ deriving (Show)++newtype GetTimestampResult = GetTimestampResult Timestamp+ deriving (Show)++newtype PutResult = PutResult Bool+ deriving (Eq, Show)++data PutResultPlus = PutResultPlus Bool [B64UUID Plus]+ deriving (Show)++data PutOffsetResult+ = PutOffsetResult Offset+ | PutOffsetResultAlreadyHave+ deriving (Show)++data PutOffsetResultPlus + = PutOffsetResultPlus Offset+ | PutOffsetResultAlreadyHavePlus [B64UUID Plus]+ deriving (Show, Generic, NFData)++newtype Offset = Offset P2P.Offset+ deriving (Show, Generic, NFData)++newtype Timestamp = Timestamp MonotonicTimestamp+ deriving (Show)++data LockResult = LockResult Bool (Maybe LockID)+ deriving (Show, Generic, NFData)++newtype UnlockRequest = UnlockRequest Bool+ deriving (Show, Generic, NFData)++-- Not using servant's built-in basic authentication support,+-- because whether authentication is needed depends on server+-- configuration.+data Auth = Auth B.ByteString B.ByteString+ deriving (Show, Generic, NFData, Eq, Ord)++#ifdef WITH_SERVANT++instance ToHttpApiData Auth where+ toHeader (Auth u p) = "Basic " <> B64.encode (u <> ":" <> p)+#if MIN_VERSION_text(2,0,0)+ toUrlPiece = TE.decodeUtf8Lenient . toHeader+#else+ toUrlPiece = TE.decodeUtf8With (\_ _ -> Just '\xfffd') . toHeader+#endif++instance FromHttpApiData Auth where+ parseHeader h =+ let (b, rest) = B.break (isSpace . chr . fromIntegral) h+ in if map toLower (decodeBS b) == "basic"+ then case B64.decode (B.dropWhile (isSpace . chr . fromIntegral) rest) of+ Right v -> case B.split (fromIntegral (ord ':')) v of+ (u:ps) -> Right $+ Auth u (B.intercalate ":" ps)+ _ -> bad+ Left _ -> bad+ else bad+ where+ bad = Left "invalid basic auth header"+ parseUrlPiece = parseHeader . encodeBS . T.unpack++newtype ConnectionKeepAlive = ConnectionKeepAlive T.Text++connectionKeepAlive :: ConnectionKeepAlive+connectionKeepAlive = ConnectionKeepAlive "Keep-Alive"++newtype KeepAlive = KeepAlive T.Text++keepAlive :: KeepAlive+keepAlive = KeepAlive "timeout=1200"++instance ToHttpApiData ConnectionKeepAlive where+ toUrlPiece (ConnectionKeepAlive t) = t++instance FromHttpApiData ConnectionKeepAlive where+ parseUrlPiece = Right . ConnectionKeepAlive++instance ToHttpApiData KeepAlive where+ toUrlPiece (KeepAlive t) = t++instance FromHttpApiData KeepAlive where+ parseUrlPiece = Right . KeepAlive++instance ToHttpApiData V3 where toUrlPiece _ = "v3"+instance ToHttpApiData V2 where toUrlPiece _ = "v2"+instance ToHttpApiData V1 where toUrlPiece _ = "v1"+instance ToHttpApiData V0 where toUrlPiece _ = "v0"++instance FromHttpApiData V3 where parseUrlPiece = parseAPIVersion V3 "v3"+instance FromHttpApiData V2 where parseUrlPiece = parseAPIVersion V2 "v2"+instance FromHttpApiData V1 where parseUrlPiece = parseAPIVersion V1 "v1"+instance FromHttpApiData V0 where parseUrlPiece = parseAPIVersion V0 "v0"++parseAPIVersion :: v -> T.Text -> T.Text -> Either T.Text v+parseAPIVersion v need t+ | t == need = Right v+ | otherwise = Left "bad version"++instance ToHttpApiData B64Key where+ toUrlPiece (B64Key k) = encodeB64Text (serializeKey' k)++instance FromHttpApiData B64Key where+ parseUrlPiece t = case decodeB64Text t of+ Right b -> maybe (Left "key parse error") (Right . B64Key)+ (deserializeKey' b)+ Left err -> Left err++instance ToHttpApiData (B64UUID t) where+ toUrlPiece (B64UUID u) = encodeB64Text (fromUUID u)++instance FromHttpApiData (B64UUID t) where+ parseUrlPiece t = case decodeB64Text t of+ Right b -> case toUUID b of+ u@(UUID _) -> Right (B64UUID u)+ NoUUID -> Left "empty UUID"+ Left err -> Left err++instance ToHttpApiData B64FilePath where+ toUrlPiece (B64FilePath f) = encodeB64Text f++instance FromHttpApiData B64FilePath where+ parseUrlPiece t = case decodeB64Text t of+ Right b -> Right (B64FilePath b)+ Left err -> Left err++instance ToHttpApiData Offset where+ toUrlPiece (Offset (P2P.Offset n)) = T.pack (show n)++instance FromHttpApiData Offset where+ parseUrlPiece t = case readMaybe (T.unpack t) of+ Nothing -> Left "offset parse error"+ Just n -> Right (Offset (P2P.Offset n))++instance ToHttpApiData Timestamp where+ toUrlPiece (Timestamp (MonotonicTimestamp n)) = T.pack (show n)++instance FromHttpApiData Timestamp where+ parseUrlPiece t = case readMaybe (T.unpack t) of+ Nothing -> Left "timestamp parse error"+ Just n -> Right (Timestamp (MonotonicTimestamp n))++instance ToHttpApiData DataLength where+ toUrlPiece (DataLength n) = T.pack (show n)++instance FromHttpApiData DataLength where+ parseUrlPiece t = case readMaybe (T.unpack t) of+ Nothing -> Left "X-git-annex-data-length parse error"+ Just n -> Right (DataLength n)++instance ToJSON PutResult where+ toJSON (PutResult b) =+ object ["stored" .= b]++instance FromJSON PutResult where+ parseJSON = withObject "PutResult" $ \v -> PutResult+ <$> v .: "stored"++instance ToJSON PutResultPlus where+ toJSON (PutResultPlus b us) = object+ [ "stored" .= b+ , "plusuuids" .= plusList us+ ]++instance FromJSON PutResultPlus where+ parseJSON = withObject "PutResultPlus" $ \v -> PutResultPlus+ <$> v .: "stored"+ <*> v .: "plusuuids"++instance ToJSON CheckPresentResult where+ toJSON (CheckPresentResult b) = object+ ["present" .= b]++instance FromJSON CheckPresentResult where+ parseJSON = withObject "CheckPresentResult" $ \v -> CheckPresentResult+ <$> v .: "present"++instance ToJSON RemoveResult where+ toJSON (RemoveResult b) = object+ ["removed" .= b]++instance FromJSON RemoveResult where+ parseJSON = withObject "RemoveResult" $ \v -> RemoveResult+ <$> v .: "removed"++instance ToJSON RemoveResultPlus where+ toJSON (RemoveResultPlus b us) = object+ [ "removed" .= b+ , "plusuuids" .= plusList us+ ]++instance FromJSON RemoveResultPlus where+ parseJSON = withObject "RemoveResultPlus" $ \v -> RemoveResultPlus+ <$> v .: "removed"+ <*> v .: "plusuuids"++instance ToJSON GetTimestampResult where+ toJSON (GetTimestampResult (Timestamp (MonotonicTimestamp t))) = object+ ["timestamp" .= t]++instance FromJSON GetTimestampResult where+ parseJSON = withObject "GetTimestampResult" $ \v ->+ GetTimestampResult . Timestamp . MonotonicTimestamp+ <$> v .: "timestamp"++instance ToJSON PutOffsetResult where+ toJSON (PutOffsetResult (Offset (P2P.Offset o))) = object+ ["offset" .= o]+ toJSON PutOffsetResultAlreadyHave = object+ ["alreadyhave" .= True]++instance FromJSON PutOffsetResult where+ parseJSON = withObject "PutOffsetResult" $ \v ->+ (PutOffsetResult+ <$> (Offset . P2P.Offset <$> v .: "offset"))+ <|> (mkalreadyhave+ <$> (v .: "alreadyhave"))+ where+ mkalreadyhave :: Bool -> PutOffsetResult+ mkalreadyhave _ = PutOffsetResultAlreadyHave++instance ToJSON PutOffsetResultPlus where+ toJSON (PutOffsetResultPlus (Offset (P2P.Offset o))) = object+ [ "offset" .= o ]+ toJSON (PutOffsetResultAlreadyHavePlus us) = object+ [ "alreadyhave" .= True+ , "plusuuids" .= plusList us+ ]++instance FromJSON PutOffsetResultPlus where+ parseJSON = withObject "PutOffsetResultPlus" $ \v ->+ (PutOffsetResultPlus+ <$> (Offset . P2P.Offset <$> v .: "offset"))+ <|> (mkalreadyhave+ <$> (v .: "alreadyhave")+ <*> (v .: "plusuuids"))+ where+ mkalreadyhave :: Bool -> [B64UUID Plus] -> PutOffsetResultPlus+ mkalreadyhave _ us = PutOffsetResultAlreadyHavePlus us++instance FromJSON (B64UUID t) where+ parseJSON (String t) = case decodeB64Text t of+ Right s -> pure (B64UUID (toUUID s))+ Left _ -> mempty+ parseJSON _ = mempty++instance ToJSON LockResult where+ toJSON (LockResult v (Just (B64UUID lck))) = object+ [ "locked" .= v+ , "lockid" .= encodeB64Text (fromUUID lck)+ ]+ toJSON (LockResult v Nothing) = object+ [ "locked" .= v+ ]++instance FromJSON LockResult where+ parseJSON = withObject "LockResult" $ \v -> LockResult+ <$> v .: "locked"+ <*> v .:? "lockid"++instance ToJSON UnlockRequest where+ toJSON (UnlockRequest v) = object+ ["unlock" .= v]++instance FromJSON UnlockRequest where+ parseJSON = withObject "UnlockRequest" $ \v -> UnlockRequest+ <$> v .: "unlock"++plusList :: [B64UUID Plus] -> [String]+plusList = map (\(B64UUID u) -> fromUUID u)++class PlusClass plus unplus where+ dePlus :: plus -> unplus+ plus :: unplus -> plus++instance PlusClass RemoveResultPlus RemoveResult where+ dePlus (RemoveResultPlus b _) = RemoveResult b+ plus (RemoveResult b) = RemoveResultPlus b mempty++instance PlusClass PutResultPlus PutResult where+ dePlus (PutResultPlus b _) = PutResult b+ plus (PutResult b) = PutResultPlus b mempty++instance PlusClass PutOffsetResultPlus PutOffsetResult where+ dePlus (PutOffsetResultPlus o) = PutOffsetResult o+ dePlus (PutOffsetResultAlreadyHavePlus _) = PutOffsetResultAlreadyHave+ plus (PutOffsetResult o) = PutOffsetResultPlus o+ plus PutOffsetResultAlreadyHave = PutOffsetResultAlreadyHavePlus []++#endif
+ P2P/Http/Url.hs view
@@ -0,0 +1,80 @@+{- P2P protocol over HTTP, urls+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - Licensed under the GNU AGPL version 3 or higher.+ -}++{-# LANGUAGE CPP #-}++module P2P.Http.Url where++import Data.List+import Network.URI+import System.FilePath.Posix as P+#ifdef WITH_SERVANT+import Servant.Client (BaseUrl(..), Scheme(..))+import Text.Read+#endif++defaultP2PHttpProtocolPort :: Int+defaultP2PHttpProtocolPort = 9417 -- Git protocol is 9418++isP2PHttpProtocolUrl :: String -> Bool+isP2PHttpProtocolUrl s = + "annex+http://" `isPrefixOf` s ||+ "annex+https://" `isPrefixOf` s++data P2PHttpUrl = P2PHttpUrl+ { p2pHttpUrlString :: String+#ifdef WITH_SERVANT+ , p2pHttpBaseUrl :: BaseUrl+#endif+ }+ deriving (Show)++parseP2PHttpUrl :: String -> Maybe P2PHttpUrl+parseP2PHttpUrl us+ | isP2PHttpProtocolUrl us = case parseURI (drop prefixlen us) of+ Nothing -> Nothing+ Just u ->+#ifdef WITH_SERVANT+ case uriScheme u of+ "http:" -> mkbaseurl Http u+ "https:" -> mkbaseurl Https u+ _ -> Nothing+#else+ Just $ P2PHttpUrl us+#endif+ | otherwise = Nothing+ where+ prefixlen = length "annex+"++#ifdef WITH_SERVANT+ mkbaseurl s u = do+ auth <- uriAuthority u+ port <- if null (uriPort auth)+ then Just defaultP2PHttpProtocolPort+ else readMaybe (dropWhile (== ':') (uriPort auth))+ return $ P2PHttpUrl us $ BaseUrl+ { baseUrlScheme = s+ , baseUrlHost = uriRegName auth+ , baseUrlPath = basepath u+ , baseUrlPort = port+ }+ + -- The servant server uses urls that start with "/git-annex/",+ -- and so the servant client adds that to the base url. So remove+ -- it from the url that the user provided. However, it may not be+ -- present, eg if some other server is speaking the git-annex+ -- protocol. The UUID is also removed from the end of the url.+ basepath u = case reverse $ P.splitDirectories (uriPath u) of+ ("git-annex":"/":rest) -> P.joinPath (reverse rest)+ rest -> P.joinPath (reverse rest)+#endif++unavailableP2PHttpUrl :: P2PHttpUrl -> P2PHttpUrl+unavailableP2PHttpUrl p = p+#ifdef WITH_SERVANT+ { p2pHttpBaseUrl = (p2pHttpBaseUrl p) { baseUrlHost = "!dne!" } }+#endif
P2P/IO.hs view
@@ -25,6 +25,7 @@ , describeProtoFailure , runNetProto , runNet+ , signalFullyConsumedByteString ) where import Common@@ -38,6 +39,7 @@ import Utility.Tor import Utility.FileMode import Utility.Debug+import Utility.MonotonicClock import Types.UUID import Annex.ChangedRefs import qualified Utility.RawFilePath as R@@ -61,6 +63,7 @@ = ProtoFailureMessage String | ProtoFailureException SomeException | ProtoFailureIOError IOError+ deriving (Show) describeProtoFailure :: ProtoFailure -> String describeProtoFailure (ProtoFailureMessage s) = s@@ -78,8 +81,18 @@ data P2PHandle = P2PHandle Handle- | P2PHandleTMVar (TMVar (Either L.ByteString Message)) (TMVar ())+ | P2PHandleTMVar+ (TMVar (Either L.ByteString Message))+ (Maybe (TMVar ()))+ (TMVar ()) +signalFullyConsumedByteString :: P2PHandle -> IO ()+signalFullyConsumedByteString (P2PHandle _) = return ()+signalFullyConsumedByteString (P2PHandleTMVar _ Nothing _) = return () +signalFullyConsumedByteString (P2PHandleTMVar _ (Just waitv) closedv) = + atomically $ putTMVar waitv ()+ `orElse` readTMVar closedv+ data P2PConnection = P2PConnection { connRepo :: Maybe Repo , connCheckAuth :: (AuthToken -> Bool)@@ -90,6 +103,7 @@ -- Identifier for a connection, only used for debugging. newtype ConnIdent = ConnIdent (Maybe String)+ deriving (Show) data ClosableConnection conn = OpenConnection conn@@ -137,7 +151,8 @@ closehandle (connOhdl conn) where closehandle (P2PHandle h) = hClose h- closehandle (P2PHandleTMVar _ _) = return ()+ closehandle (P2PHandleTMVar _ _ closedv) = + atomically $ void $ tryPutTMVar closedv () -- Serves the protocol on a unix socket. --@@ -187,11 +202,6 @@ go (Free (Local _)) = return $ Left $ ProtoFailureMessage "unexpected annex operation attempted" -data P2PTMVarException = P2PTMVarException String- deriving (Show)--instance Exception P2PTMVarException- -- Interpreter of the Net part of Proto. -- -- An interpreter of Proto has to be provided, to handle the rest of Proto@@ -205,18 +215,15 @@ P2PHandle h -> tryNonAsync $ do hPutStrLn h $ unwords (formatMessage m) hFlush h- P2PHandleTMVar mv _ ->- ifM (atomically (tryPutTMVar mv (Right m)))- ( return $ Right ()- , return $ Left $ toException $- P2PTMVarException "TMVar left full"- )+ P2PHandleTMVar mv _ closedv -> tryNonAsync $+ atomically $ putTMVar mv (Right m)+ `orElse` readTMVar closedv case v of Left e -> return $ Left $ ProtoFailureException e Right () -> runner next ReceiveMessage next -> let protoerr = return $ Left $- ProtoFailureMessage "protocol error 1"+ ProtoFailureMessage "protocol error" gotmessage m = do liftIO $ debugMessage conn "P2P <" m runner (next (Just m))@@ -229,10 +236,13 @@ Right (Just l) -> case parseMessage l of Just m -> gotmessage m Nothing -> runner (next Nothing)- P2PHandleTMVar mv _ -> - liftIO (atomically (takeTMVar mv)) >>= \case- Right m -> gotmessage m- Left _b -> protoerr+ P2PHandleTMVar mv _ closedv -> do+ let recv = (Just <$> takeTMVar mv)+ `orElse` (readTMVar closedv >> return Nothing)+ liftIO (atomically recv) >>= \case+ Just (Right m) -> gotmessage m+ Just (Left _b) -> protoerr+ Nothing -> runner (next Nothing) SendBytes len b p next -> case connOhdl conn of P2PHandle h -> do@@ -245,11 +255,16 @@ Right False -> return $ Left $ ProtoFailureMessage "short data write" Left e -> return $ Left $ ProtoFailureException e- P2PHandleTMVar mv waitv -> do+ P2PHandleTMVar mv waitv closedv -> do liftIO $ atomically $ putTMVar mv (Left b)- -- Wait for the whole bytestring to be- -- processed. Necessary due to lazyiness.- liftIO $ atomically $ takeTMVar waitv+ `orElse` readTMVar closedv+ -- Wait for the whole bytestring to+ -- be processed.+ case waitv of+ Nothing -> noop+ Just v -> liftIO $ atomically $+ takeTMVar v+ `orElse` readTMVar closedv runner next ReceiveBytes len p next -> case connIhdl conn of@@ -259,11 +274,15 @@ Right b -> runner (next b) Left e -> return $ Left $ ProtoFailureException e- P2PHandleTMVar mv _ ->- liftIO (atomically (takeTMVar mv)) >>= \case- Left b -> runner (next b)- Right _ -> return $ Left $- ProtoFailureMessage "protocol error 2"+ P2PHandleTMVar mv _ closedv -> do+ let recv = (Just <$> takeTMVar mv)+ `orElse` (readTMVar closedv >> return Nothing)+ liftIO (atomically recv) >>= \case+ Just (Left b) -> runner (next b)+ Just (Right _) -> return $ Left $+ ProtoFailureMessage "protocol error"+ Nothing -> return $ Left $+ ProtoFailureMessage "connection closed" CheckAuthToken _u t next -> do let authed = connCheckAuth conn t runner (next authed)@@ -282,6 +301,8 @@ runner next GetProtocolVersion next -> liftIO (readTVarIO versiontvar) >>= runner . next+ GetMonotonicTimestamp next ->+ liftIO currentMonotonicTimestamp >>= runner . next where -- This is only used for running Net actions when relaying, -- so it's ok to use runNetProto, despite it not supporting@@ -314,12 +335,16 @@ -- Must avoid sending too many bytes as it would confuse the other end. -- This is easily dealt with by truncating it. --+-- However, the whole ByteString will be evaluated here, even if+-- the end of it does not get sent.+-- -- If too few bytes are sent, the only option is to give up on this -- connection. False is returned to indicate this problem. sendExactly :: Len -> L.ByteString -> Handle -> MeterUpdate -> IO Bool sendExactly (Len n) b h p = do- sent <- meteredWrite' p (B.hPut h) (L.take (fromIntegral n) b)- return (fromBytesProcessed sent == n)+ let (x, y) = L.splitAt (fromIntegral n) b+ sent <- meteredWrite' p (B.hPut h) x+ L.length y `seq` return (fromBytesProcessed sent == n) receiveExactly :: Len -> Handle -> MeterUpdate -> IO L.ByteString receiveExactly (Len n) h p = hGetMetered h (Just n) p
P2P/Protocol.hs view
@@ -26,8 +26,10 @@ import Utility.PartialPrelude import Utility.Metered import Utility.FileSystemEncoding+import Utility.MonotonicClock import Git.FilePath import Annex.ChangedRefs (ChangedRefs)+import Types.NumCopies import Control.Monad import Control.Monad.Free@@ -40,11 +42,14 @@ import qualified Data.ByteString.Lazy as L import qualified Data.Set as S import Data.Char+import Data.Maybe+import Data.Time.Clock.POSIX import Control.Applicative+import Control.DeepSeq import Prelude newtype Offset = Offset Integer- deriving (Show)+ deriving (Show, Eq, NFData, Num, Real, Ord, Enum, Integral) newtype Len = Len Integer deriving (Show)@@ -56,8 +61,17 @@ defaultProtocolVersion = ProtocolVersion 0 maxProtocolVersion :: ProtocolVersion-maxProtocolVersion = ProtocolVersion 2+maxProtocolVersion = ProtocolVersion 3 +-- In order from newest to oldest.+allProtocolVersions :: [ProtocolVersion]+allProtocolVersions =+ [ ProtocolVersion 3+ , ProtocolVersion 2+ , ProtocolVersion 1+ , ProtocolVersion 0+ ] + newtype ProtoAssociatedFile = ProtoAssociatedFile AssociatedFile deriving (Show) @@ -86,6 +100,8 @@ | LOCKCONTENT Key | UNLOCKCONTENT | REMOVE Key+ | REMOVE_BEFORE MonotonicTimestamp Key+ | GETTIMESTAMP | GET Offset ProtoAssociatedFile Key | PUT ProtoAssociatedFile Key | PUT_FROM Offset@@ -98,6 +114,7 @@ | BYPASS Bypass | DATA Len -- followed by bytes of data | VALIDITY Validity+ | TIMESTAMP MonotonicTimestamp | ERROR String deriving (Show) @@ -114,6 +131,8 @@ formatMessage (LOCKCONTENT key) = ["LOCKCONTENT", Proto.serialize key] formatMessage UNLOCKCONTENT = ["UNLOCKCONTENT"] formatMessage (REMOVE key) = ["REMOVE", Proto.serialize key]+ formatMessage (REMOVE_BEFORE ts key) = ["REMOVE-BEFORE", Proto.serialize ts, Proto.serialize key]+ formatMessage GETTIMESTAMP = ["GETTIMESTAMP"] formatMessage (GET offset af key) = ["GET", Proto.serialize offset, Proto.serialize af, Proto.serialize key] formatMessage (PUT af key) = ["PUT", Proto.serialize af, Proto.serialize key] formatMessage (PUT_FROM offset) = ["PUT-FROM", Proto.serialize offset]@@ -124,9 +143,10 @@ formatMessage FAILURE = ["FAILURE"] formatMessage (FAILURE_PLUS uuids) = ("FAILURE-PLUS":map Proto.serialize uuids) formatMessage (BYPASS (Bypass uuids)) = ("BYPASS":map Proto.serialize (S.toList uuids))+ formatMessage (DATA len) = ["DATA", Proto.serialize len] formatMessage (VALIDITY Valid) = ["VALID"] formatMessage (VALIDITY Invalid) = ["INVALID"]- formatMessage (DATA len) = ["DATA", Proto.serialize len]+ formatMessage (TIMESTAMP ts) = ["TIMESTAMP", Proto.serialize ts] formatMessage (ERROR err) = ["ERROR", Proto.serialize err] instance Proto.Receivable Message where@@ -142,6 +162,8 @@ parseCommand "LOCKCONTENT" = Proto.parse1 LOCKCONTENT parseCommand "UNLOCKCONTENT" = Proto.parse0 UNLOCKCONTENT parseCommand "REMOVE" = Proto.parse1 REMOVE+ parseCommand "REMOVE-BEFORE" = Proto.parse2 REMOVE_BEFORE+ parseCommand "GETTIMESTAMP" = Proto.parse0 GETTIMESTAMP parseCommand "GET" = Proto.parse3 GET parseCommand "PUT" = Proto.parse2 PUT parseCommand "PUT-FROM" = Proto.parse1 PUT_FROM@@ -153,9 +175,10 @@ parseCommand "FAILURE-PLUS" = Proto.parseList FAILURE_PLUS parseCommand "BYPASS" = Proto.parseList (BYPASS . Bypass . S.fromList) parseCommand "DATA" = Proto.parse1 DATA- parseCommand "ERROR" = Proto.parse1 ERROR parseCommand "VALID" = Proto.parse0 (VALIDITY Valid) parseCommand "INVALID" = Proto.parse0 (VALIDITY Invalid)+ parseCommand "TIMESTAMP" = Proto.parse1 TIMESTAMP+ parseCommand "ERROR" = Proto.parse1 ERROR parseCommand _ = Proto.parseFail instance Proto.Serializable ProtocolVersion where@@ -170,6 +193,10 @@ serialize (Len n) = show n deserialize = Len <$$> readish +instance Proto.Serializable MonotonicTimestamp where+ serialize (MonotonicTimestamp n) = show n+ deserialize = MonotonicTimestamp <$$> readish+ instance Proto.Serializable Service where serialize UploadPack = "git-upload-pack" serialize ReceivePack = "git-receive-pack"@@ -234,9 +261,9 @@ -- ^ Sends exactly Len bytes of data. (Any more or less will -- confuse the receiver.) | ReceiveBytes Len MeterUpdate (L.ByteString -> c)- -- ^ Lazily reads bytes from peer. Stops once Len are read,- -- or if connection is lost, and in either case returns the bytes- -- that were read. This allows resuming interrupted transfers.+ -- ^ Streams bytes from peer. Stops once Len are read,+ -- or if connection is lost. This allows resuming+ -- interrupted transfers. | CheckAuthToken UUID AuthToken (Bool -> c) | RelayService Service c -- ^ Runs a service, relays its output to the peer, and data@@ -249,6 +276,7 @@ | SetProtocolVersion ProtocolVersion c --- ^ Called when a new protocol version has been negotiated. | GetProtocolVersion (ProtocolVersion -> c)+ | GetMonotonicTimestamp (MonotonicTimestamp -> c) deriving (Functor) type Net = Free NetF@@ -291,12 +319,18 @@ -- content been transferred. | StoreContentTo FilePath (Maybe IncrementalVerifier) Offset Len (Proto L.ByteString) (Proto (Maybe Validity)) ((Bool, Verification) -> c) -- ^ Like StoreContent, but stores the content to a temp file.+ | SendContentWith (L.ByteString -> Annex (Maybe Validity -> Annex Bool)) (Proto L.ByteString) (Proto (Maybe Validity)) (Bool -> c)+ -- ^ Reads content from the Proto L.ByteString and sends it to the+ -- callback. The callback must consume the whole lazy ByteString,+ -- before it returns a validity checker. | SetPresent Key UUID c | CheckContentPresent Key (Bool -> c) -- ^ Checks if the whole content of the key is locally present.- | RemoveContent Key (Bool -> c)+ | RemoveContent Key (Maybe MonotonicTimestamp) (Bool -> c) -- ^ If the content is not present, still succeeds. -- May fail if not enough copies to safely drop, etc.+ -- After locking the content for removal, checks if it's later+ -- than the MonotonicTimestamp, and fails. | TryLockContent Key (Bool -> Proto ()) c -- ^ Try to lock the content of a key, preventing it -- from being deleted, while running the provided protocol@@ -309,6 +343,8 @@ -- not known until the data is being received. | RunValidityCheck (Annex Validity) (Validity -> c) -- ^ Runs a deferred validity check.+ | GetLocalCurrentTime (POSIXTime -> c)+ -- ^ Gets the local time. deriving (Functor) type Local = Free LocalF@@ -341,7 +377,7 @@ case r of Just (VERSION v) -> net $ setProtocolVersion v -- Old server doesn't know about the VERSION command.- Just (ERROR _) -> return ()+ Just (ERROR _) -> net $ setProtocolVersion (ProtocolVersion 0) _ -> net $ sendMessage (ERROR "expected VERSION") sendBypass :: Bypass -> Proto ()@@ -378,11 +414,67 @@ cleanup True = runproto () $ net $ sendMessage UNLOCKCONTENT cleanup False = return () -remove :: Key -> Proto (Either String Bool, Maybe [UUID])-remove key = do- net $ sendMessage (REMOVE key)- checkSuccessFailurePlus+remove :: Maybe SafeDropProof -> Key -> Proto (Either String Bool, Maybe [UUID])+remove proof key = + case safeDropProofEndTime =<< proof of+ Nothing -> removeanytime+ Just endtime -> do+ ver <- net getProtocolVersion+ if ver >= ProtocolVersion 3+ then removeBefore endtime key+ -- Peer is too old to support REMOVE-BEFORE+ else removeanytime+ where+ removeanytime = do+ net $ sendMessage (REMOVE key)+ checkSuccessFailurePlus +getTimestamp :: Proto (Either String MonotonicTimestamp)+getTimestamp = do+ net $ sendMessage GETTIMESTAMP+ net receiveMessage >>= \case+ Just (TIMESTAMP ts) -> return (Right ts)+ Just (ERROR err) -> return (Left err)+ _ -> do+ net $ sendMessage (ERROR "expected TIMESTAMP")+ return (Left "protocol error")++removeBefore :: POSIXTime -> Key -> Proto (Either String Bool, Maybe [UUID])+removeBefore endtime key = getTimestamp >>= \case+ Right remotetime -> + canRemoveBefore endtime remotetime (local getLocalCurrentTime) >>= \case+ Just remoteendtime -> + removeBeforeRemoteEndTime remoteendtime key+ Nothing ->+ return (Right False, Nothing)+ Left err -> return (Left err, Nothing)++{- The endtime is the last local time at which the key can be removed.+ - To tell the remote how long it has to remove the key, get its current+ - timestamp, and add to it the number of seconds from the current local+ - time until the endtime.+ -+ - Order of retrieving timestamps matters. Getting the local time after the+ - remote timestamp means that, if there is some delay in getting the+ - response from the remote, that is reflected in the local time, and so+ - reduces the allowed time.+ -}+canRemoveBefore :: Monad m => POSIXTime -> MonotonicTimestamp -> m POSIXTime -> m (Maybe MonotonicTimestamp)+canRemoveBefore endtime remotetime getlocaltime = do+ localtime <- getlocaltime+ let timeleft = endtime - localtime+ let timeleft' = MonotonicTimestamp (floor timeleft)+ let remoteendtime = remotetime + timeleft'+ return $ if timeleft <= 0+ then Nothing+ else Just remoteendtime++removeBeforeRemoteEndTime :: MonotonicTimestamp -> Key -> Proto (Either String Bool, Maybe [UUID])+removeBeforeRemoteEndTime remoteendtime key = do+ net $ sendMessage $+ REMOVE_BEFORE remoteendtime key+ checkSuccessFailurePlus + get :: FilePath -> Key -> Maybe IncrementalVerifier -> AssociatedFile -> Meter -> MeterUpdate -> Proto (Bool, Verification) get dest key iv af m p = receiveContent (Just m) p sizer storer $ \offset ->@@ -392,17 +484,39 @@ storer = storeContentTo dest iv put :: Key -> AssociatedFile -> MeterUpdate -> Proto (Maybe [UUID])-put key af p = do+put key af p = put' key af $ \offset ->+ sendContent key af Nothing offset p++put' :: Key -> AssociatedFile -> (Offset -> Proto (Maybe [UUID])) -> Proto (Maybe [UUID])+put' key af sender = do net $ sendMessage (PUT (ProtoAssociatedFile af) key) r <- net receiveMessage case r of- Just (PUT_FROM offset) -> sendContent key af Nothing offset p+ Just (PUT_FROM offset) -> sender offset Just ALREADY_HAVE -> return (Just []) Just (ALREADY_HAVE_PLUS uuids) -> return (Just uuids) _ -> do net $ sendMessage (ERROR "expected PUT_FROM or ALREADY_HAVE") return Nothing +-- The protocol does not have a way to get the PUT offset+-- without sending DATA, so send an empty bytestring and indicate+-- it is not valid.+getPutOffset :: Key -> AssociatedFile -> Proto (Either [UUID] Offset)+getPutOffset key af = do+ net $ sendMessage (PUT (ProtoAssociatedFile af) key)+ r <- net receiveMessage+ case r of+ Just (PUT_FROM offset) -> do+ void $ sendContent' nullMeterUpdate (Len 0) L.empty $+ return Invalid+ return (Right offset)+ Just ALREADY_HAVE -> return (Left [])+ Just (ALREADY_HAVE_PLUS uuids) -> return (Left uuids)+ _ -> do+ net $ sendMessage (ERROR "expected PUT_FROM or ALREADY_HAVE")+ return (Left [])+ data ServerHandler a = ServerGot a | ServerContinue@@ -410,7 +524,14 @@ -- Server loop, getting messages from the client and handling them serverLoop :: (Message -> Proto (ServerHandler a)) -> Proto (Maybe a)-serverLoop a = do+serverLoop a = serveOneMessage a serverLoop++-- Get one message from the client and handle it.+serveOneMessage+ :: (Message -> Proto (ServerHandler a))+ -> ((Message -> Proto (ServerHandler a)) -> Proto (Maybe a)) + -> Proto (Maybe a)+serveOneMessage a cont = do mcmd <- net receiveMessage case mcmd of -- When the client sends ERROR to the server, the server@@ -418,16 +539,16 @@ -- is in, and so not possible to recover. Just (ERROR _) -> return Nothing -- When the client sends an unparsable message, the server- -- responds with an error message, and loops. This allows+ -- responds with an error message, and continues. This allows -- expanding the protocol with new messages. Nothing -> do net $ sendMessage (ERROR "unknown command")- serverLoop a+ cont a Just cmd -> do v <- a cmd case v of ServerGot r -> return (Just r)- ServerContinue -> serverLoop a+ ServerContinue -> cont a -- If the client sends an unexpected message, -- the server will respond with ERROR, and -- always continues processing messages.@@ -439,7 +560,7 @@ -- support some new feature, and fall back. ServerUnexpected -> do net $ sendMessage (ERROR "unexpected command")- serverLoop a+ cont a -- | Serve the protocol, with an unauthenticated peer. Once the peer -- successfully authenticates, returns their UUID.@@ -464,11 +585,22 @@ -- ^ Allow reading, and storing new objects, but not deleting objects. | ServeReadWrite -- ^ Full read and write access.- deriving (Eq, Ord)+ deriving (Show, Eq, Ord) -- | Serve the protocol, with a peer that has authenticated. serveAuthed :: ServerMode -> UUID -> Proto ()-serveAuthed servermode myuuid = void $ serverLoop handler+serveAuthed servermode myuuid = void $ serverLoop $+ serverHandler servermode myuuid++-- | Serve a single command in the protocol, the same as serveAuthed,+-- but without looping to handle the next command.+serveOneCommandAuthed :: ServerMode -> UUID -> Proto ()+serveOneCommandAuthed servermode myuuid = fromMaybe () <$>+ serveOneMessage (serverHandler servermode myuuid) + (const $ pure Nothing)++serverHandler :: ServerMode -> UUID -> Message -> Proto (ServerHandler ())+serverHandler servermode myuuid = handler where handler (VERSION theirversion) = do let v = min theirversion maxProtocolVersion@@ -488,13 +620,13 @@ sendSuccess =<< local (checkContentPresent key) return ServerContinue handler (REMOVE key) =- checkREMOVEServerMode servermode $ \case- Nothing -> do- sendSuccess =<< local (removeContent key)- return ServerContinue - Just notallowed -> do- notallowed- return ServerContinue+ handleremove key Nothing+ handler (REMOVE_BEFORE ts key) =+ handleremove key (Just ts)+ handler GETTIMESTAMP = do+ ts <- net getMonotonicTimestamp+ net $ sendMessage $ TIMESTAMP ts+ return ServerContinue handler (PUT (ProtoAssociatedFile af) key) = checkPUTServerMode servermode $ \case Nothing -> handleput af key@@ -536,6 +668,15 @@ when (observeBool v) $ local $ setPresent key myuuid return ServerContinue+ + handleremove key mts =+ checkREMOVEServerMode servermode $ \case+ Nothing -> do+ sendSuccess =<< local (removeContent key mts)+ return ServerContinue + Just notallowed -> do+ notallowed+ return ServerContinue sendReadOnlyError :: Proto () sendReadOnlyError = net $ sendMessage $@@ -580,16 +721,21 @@ else local $ readContent key af o offset $ sender (Len len) -- Content not available to send. Indicate this by sending- -- empty data and indlicate it's invalid.+ -- empty data and indicate it's invalid. go Nothing = sender (Len 0) L.empty (return Invalid)- sender len content validitycheck = do- let p' = offsetMeterUpdate p (toBytesProcessed n)- net $ sendMessage (DATA len)- net $ sendBytes len content p'- ver <- net getProtocolVersion- when (ver >= ProtocolVersion 1) $- net . sendMessage . VALIDITY =<< validitycheck- checkSuccessPlus++ sender = sendContent' p'+ + p' = offsetMeterUpdate p (toBytesProcessed n)++sendContent' :: MeterUpdate -> Len -> L.ByteString -> Proto Validity -> Proto (Maybe [UUID])+sendContent' p len content validitycheck = do+ net $ sendMessage (DATA len)+ net $ sendBytes len content p+ ver <- net getProtocolVersion+ when (ver >= ProtocolVersion 1) $+ net . sendMessage . VALIDITY =<< validitycheck+ checkSuccessPlus receiveContent :: Observable t
P2P/Proxy.hs view
@@ -15,6 +15,7 @@ import P2P.Protocol import P2P.IO import Utility.Metered+import Utility.MonotonicClock import Git.FilePath import Types.Concurrency import Annex.Concurrent@@ -26,23 +27,33 @@ import qualified Control.Concurrent.MSem as MSem import qualified Data.ByteString.Lazy as L import qualified Data.Set as S+import qualified Data.Map as M+import Data.Unique import GHC.Conc type ProtoCloser = Annex () data ClientSide = ClientSide RunState P2PConnection +newtype RemoteSideId = RemoteSideId Unique+ deriving (Eq, Ord)+ data RemoteSide = RemoteSide { remote :: Remote , remoteConnect :: Annex (Maybe (RunState, P2PConnection, ProtoCloser)) , remoteTMVar :: TMVar (RunState, P2PConnection, ProtoCloser)+ , remoteSideId :: RemoteSideId } +instance Show RemoteSide where+ show rs = show (remote rs)+ mkRemoteSide :: Remote -> Annex (Maybe (RunState, P2PConnection, ProtoCloser)) -> Annex RemoteSide mkRemoteSide r remoteconnect = RemoteSide <$> pure r <*> pure remoteconnect <*> liftIO (atomically newEmptyTMVar)+ <*> liftIO (RemoteSideId <$> newUnique) runRemoteSide :: RemoteSide -> Proto a -> Annex (Either ProtoFailure a) runRemoteSide remoteside a = @@ -68,9 +79,11 @@ data ProxySelector = ProxySelector { proxyCHECKPRESENT :: Key -> Annex (Maybe RemoteSide) , proxyLOCKCONTENT :: Key -> Annex (Maybe RemoteSide)- , proxyUNLOCKCONTENT :: Annex (Maybe RemoteSide) , proxyREMOVE :: Key -> Annex [RemoteSide] -- ^ remove from all of these remotes+ , proxyGETTIMESTAMP :: Annex [RemoteSide]+ -- ^ should send every remote that proxyREMOVE can+ -- ever return for any key , proxyGET :: Key -> Annex (Maybe RemoteSide) , proxyPUT :: AssociatedFile -> Key -> Annex [RemoteSide] -- ^ put to some/all of these remotes@@ -80,8 +93,8 @@ singleProxySelector r = ProxySelector { proxyCHECKPRESENT = const (pure (Just r)) , proxyLOCKCONTENT = const (pure (Just r))- , proxyUNLOCKCONTENT = pure (Just r) , proxyREMOVE = const (pure [r])+ , proxyGETTIMESTAMP = pure [r] , proxyGET = const (pure (Just r)) , proxyPUT = const (const (pure [r])) }@@ -178,78 +191,98 @@ -- Pass along non-BYPASS message from version 0 client. cont (Bypass S.empty, (Just othermsg)) +data ProxyState = ProxyState+ { proxyRemoteLatestTimestamps :: TVar (M.Map RemoteSideId MonotonicTimestamp)+ , proxyRemoteLatestLocalTimestamp :: TVar (Maybe MonotonicTimestamp)+ }++mkProxyState :: IO ProxyState+mkProxyState = ProxyState+ <$> newTVarIO mempty+ <*> newTVarIO Nothing++data ProxyParams = ProxyParams+ { proxyMethods :: ProxyMethods+ , proxyState :: ProxyState+ , proxyServerMode :: ServerMode+ , proxyClientSide :: ClientSide+ , proxyUUID :: UUID+ , proxySelector :: ProxySelector+ , proxyConcurrencyConfig :: ConcurrencyConfig+ , proxyClientProtocolVersion :: ProtocolVersion+ -- ^ The remote(s) may speak an earlier version, or the same+ -- version, but not a later version.+ }+ {- Proxy between the client and the remote. This picks up after - sendClientProtocolVersion. -} proxy :: Annex r- -> ProxyMethods- -> ServerMode- -> ClientSide- -> UUID- -> ProxySelector- -> ConcurrencyConfig- -> ProtocolVersion- -- ^ Protocol version being spoken between the proxy and the- -- client. When there are multiple remotes, some may speak an- -- earlier version.+ -> ProxyParams -> Maybe Message -- ^ non-VERSION message that was received from the client when -- negotiating protocol version, and has not been responded to yet -> ProtoErrorHandled r-proxy proxydone proxymethods servermode (ClientSide clientrunst clientconn) remoteuuid proxyselector concurrencyconfig (ProtocolVersion protocolversion) othermsg protoerrhandler = do+proxy proxydone proxyparams othermsg protoerrhandler = do case othermsg of Nothing -> proxynextclientmessage () Just message -> proxyclientmessage (Just message) where- client = liftIO . runNetProto clientrunst clientconn+ proxyclientmessage Nothing = proxydone+ proxyclientmessage (Just message) = proxyRequest+ proxydone proxyparams proxynextclientmessage+ message protoerrhandler proxynextclientmessage () = protoerrhandler proxyclientmessage $ client (net receiveMessage)-- servermodechecker c a = c servermode $ \case- Nothing -> a- Just notallowed -> - protoerrhandler proxynextclientmessage $- client notallowed+ + client = liftIO . runNetProto clientrunst clientconn+ + ClientSide clientrunst clientconn = proxyClientSide proxyparams - proxyclientmessage Nothing = proxydone- proxyclientmessage (Just message) = case message of- CHECKPRESENT k -> proxyCHECKPRESENT proxyselector k >>= \case+{- Handles proxying a single request between the client and remote. -}+proxyRequest + :: Annex r+ -> ProxyParams+ -> (() -> Annex r) -- ^ called once the request has been handled+ -> Message+ -> ProtoErrorHandled r+proxyRequest proxydone proxyparams requestcomplete requestmessage protoerrhandler =+ case requestmessage of+ CHECKPRESENT k -> proxyCHECKPRESENT (proxySelector proxyparams) k >>= \case Just remoteside -> - proxyresponse remoteside message- (const proxynextclientmessage)+ proxyresponse remoteside requestmessage+ (const requestcomplete) Nothing -> - protoerrhandler proxynextclientmessage $+ protoerrhandler requestcomplete $ client $ net $ sendMessage FAILURE- LOCKCONTENT k -> proxyLOCKCONTENT proxyselector k >>= \case+ LOCKCONTENT k -> proxyLOCKCONTENT (proxySelector proxyparams) k >>= \case Just remoteside -> - proxyresponse remoteside message - (const proxynextclientmessage)+ handleLOCKCONTENT remoteside requestmessage Nothing ->- protoerrhandler proxynextclientmessage $+ protoerrhandler requestcomplete $ client $ net $ sendMessage FAILURE- UNLOCKCONTENT -> proxyUNLOCKCONTENT proxyselector >>= \case- Just remoteside ->- proxynoresponse remoteside message- proxynextclientmessage- Nothing -> proxynextclientmessage () REMOVE k -> do- remotesides <- proxyREMOVE proxyselector k+ remotesides <- proxyREMOVE (proxySelector proxyparams) k servermodechecker checkREMOVEServerMode $- handleREMOVE remotesides k message- GET _ _ k -> proxyGET proxyselector k >>= \case- Just remoteside -> handleGET remoteside message- Nothing -> - protoerrhandler proxynextclientmessage $- client $ net $ sendMessage $ - ERROR "content not present"+ handleREMOVE remotesides k requestmessage+ REMOVE_BEFORE _ k -> do+ remotesides <- proxyREMOVE (proxySelector proxyparams) k+ servermodechecker checkREMOVEServerMode $+ handleREMOVE remotesides k requestmessage+ GETTIMESTAMP -> do+ remotesides <- proxyGETTIMESTAMP (proxySelector proxyparams)+ handleGETTIMESTAMP remotesides+ GET _ _ k -> proxyGET (proxySelector proxyparams) k >>= \case+ Just remoteside -> handleGET remoteside requestmessage+ Nothing -> handleGETNoRemoteSide PUT paf k -> do af <- getassociatedfile paf- remotesides <- proxyPUT proxyselector af k+ remotesides <- proxyPUT (proxySelector proxyparams) af k servermodechecker checkPUTServerMode $- handlePUT remotesides k message- BYPASS _ -> proxynextclientmessage ()+ handlePUT remotesides k requestmessage+ BYPASS _ -> requestcomplete () -- These messages involve the git repository, not the -- annex. So they affect the git repository of the proxy, -- not the remote.@@ -268,6 +301,7 @@ FAILURE_PLUS _ -> protoerr DATA _ -> protoerr VALIDITY _ -> protoerr+ UNLOCKCONTENT -> protoerr -- If the client errors out, give up. ERROR msg -> giveup $ "client error: " ++ msg -- Messages that only the server should send.@@ -278,10 +312,21 @@ PUT_FROM _ -> protoerr ALREADY_HAVE -> protoerr ALREADY_HAVE_PLUS _ -> protoerr+ TIMESTAMP _ -> protoerr -- Early messages that the client should not send now. AUTH _ _ -> protoerr VERSION _ -> protoerr+ where+ client = liftIO . runNetProto clientrunst clientconn+ + ClientSide clientrunst clientconn = proxyClientSide proxyparams + servermodechecker c a = c (proxyServerMode proxyparams) $ \case+ Nothing -> a+ Just notallowed -> + protoerrhandler requestcomplete $+ client notallowed+ -- Send a message to the remote, send its response back to the -- client, and pass it to the continuation. proxyresponse remoteside message a = @@ -289,11 +334,6 @@ protoerrhandler (a resp) $ client $ net $ sendMessage resp - -- Send a message to the remote, that it will not respond to.- proxynoresponse remoteside message a =- protoerrhandler a $- runRemoteSide remoteside $ net $ sendMessage message- -- Send a message to the endpoint and get back its response. getresponse endpoint message handleresp = protoerrhandler (withresp handleresp) $ @@ -309,24 +349,86 @@ -- Read a message from one party, send it to the other, -- and then pass the message to the continuation. relayonemessage from to cont =- flip protoerrhandler (from $ net $ receiveMessage) $+ flip protoerrhandler (from $ net receiveMessage) $ withresp $ \message -> protoerrhandler (cont message) $ to $ net $ sendMessage message protoerr = do- _ <- client $ net $ sendMessage (ERROR "protocol error X")- giveup "protocol error M"+ _ <- client $ net $ sendMessage (ERROR "protocol error")+ giveup "protocol error" + handleLOCKCONTENT remoteside msg =+ proxyresponse remoteside msg $ \r () -> case r of+ SUCCESS -> relayonemessage client+ (runRemoteSide remoteside)+ (const requestcomplete)+ FAILURE -> requestcomplete ()+ _ -> requestcomplete ()+ + -- When there is a single remote, reply with its timestamp,+ -- to avoid needing timestamp translation.+ handleGETTIMESTAMP (remoteside:[]) = do+ liftIO $ atomically $ do+ writeTVar (proxyRemoteLatestTimestamps (proxyState proxyparams))+ mempty+ writeTVar (proxyRemoteLatestLocalTimestamp (proxyState proxyparams))+ Nothing+ proxyresponse remoteside GETTIMESTAMP+ (const requestcomplete)+ -- When there are multiple remotes, reply with our local timestamp,+ -- and do timestamp translation when sending REMOVE-FROM.+ handleGETTIMESTAMP remotesides = do+ -- Order of getting timestamps matters.+ -- Getting the local time after the time of the remotes+ -- means that if there is some delay in getting the time+ -- from a remote, that is reflected in the local time,+ -- and so reduces the allowed time.+ remotetimes <- (M.fromList . mapMaybe join) <$> getremotetimes + localtime <- liftIO currentMonotonicTimestamp+ liftIO $ atomically $ do+ writeTVar (proxyRemoteLatestTimestamps (proxyState proxyparams))+ remotetimes+ writeTVar (proxyRemoteLatestLocalTimestamp (proxyState proxyparams))+ (Just localtime)+ protoerrhandler requestcomplete $+ client $ net $ sendMessage (TIMESTAMP localtime)+ where+ getremotetimes = forMC (proxyConcurrencyConfig proxyparams) remotesides $ \r ->+ runRemoteSideOrSkipFailed r $ do+ net $ sendMessage GETTIMESTAMP+ net receiveMessage >>= return . \case+ Just (TIMESTAMP ts) ->+ Just (remoteSideId r, ts)+ _ -> Nothing++ proxyTimestamp ts _ _ Nothing = ts -- not proxying timestamps+ proxyTimestamp ts r tsm (Just correspondinglocaltime) =+ case M.lookup (remoteSideId r) tsm of+ Just oldts -> oldts + (ts - correspondinglocaltime)+ Nothing -> ts -- not reached+ handleREMOVE [] _ _ = -- When no places are provided to remove from, -- don't report a successful remote.- protoerrhandler proxynextclientmessage $+ protoerrhandler requestcomplete $ client $ net $ sendMessage FAILURE handleREMOVE remotesides k message = do- v <- forMC concurrencyconfig remotesides $ \r ->+ tsm <- liftIO $ readTVarIO $ + proxyRemoteLatestTimestamps (proxyState proxyparams)+ oldlocaltime <- liftIO $ readTVarIO $+ proxyRemoteLatestLocalTimestamp (proxyState proxyparams)+ v <- forMC (proxyConcurrencyConfig proxyparams) remotesides $ \r -> runRemoteSideOrSkipFailed r $ do- net $ sendMessage message+ case message of+ REMOVE_BEFORE ts _ -> do + v <- net getProtocolVersion+ if v < ProtocolVersion 3+ then net $ sendMessage $+ REMOVE k+ else net $ sendMessage $+ REMOVE_BEFORE (proxyTimestamp ts r tsm oldlocaltime) k+ _ -> net $ sendMessage message net receiveMessage >>= return . \case Just SUCCESS -> Just ((True, Nothing), [Remote.uuid (remote r)])@@ -341,11 +443,11 @@ _ -> Nothing let v' = map join v let us = concatMap snd $ catMaybes v'- mapM_ (\u -> removedContent proxymethods u k) us- protoerrhandler proxynextclientmessage $+ mapM_ (\u -> removedContent (proxyMethods proxyparams) u k) us+ protoerrhandler requestcomplete $ client $ net $ sendMessage $- let nonplussed = all (== remoteuuid) us - || protocolversion < 2+ let nonplussed = all (== proxyUUID proxyparams) us + || proxyClientProtocolVersion proxyparams < ProtocolVersion 2 in if all (maybe False (fst . fst)) v' then if nonplussed then SUCCESS@@ -355,19 +457,28 @@ [] -> FAILURE (err:_) -> ERROR err else FAILURE_PLUS us+ + -- Send an empty DATA and indicate it was invalid.+ handleGETNoRemoteSide = protoerrhandler requestcomplete $+ client $ net $ do+ sendMessage $ DATA (Len 0)+ sendBytes (Len 0) mempty nullMeterUpdate+ when (proxyClientProtocolVersion proxyparams /= ProtocolVersion 0) $+ sendMessage (VALIDITY Invalid)+ void $ receiveMessage handleGET remoteside message = getresponse (runRemoteSide remoteside) message $ withDATA (relayGET remoteside) $ \case- ERROR err -> protoerrhandler proxynextclientmessage $+ ERROR err -> protoerrhandler requestcomplete $ client $ net $ sendMessage (ERROR err) _ -> protoerr handlePUT (remoteside:[]) k message- | Remote.uuid (remote remoteside) == remoteuuid =+ | Remote.uuid (remote remoteside) == proxyUUID proxyparams = getresponse (runRemoteSide remoteside) message $ \resp -> case resp of- ALREADY_HAVE -> protoerrhandler proxynextclientmessage $+ ALREADY_HAVE -> protoerrhandler requestcomplete $ client $ net $ sendMessage resp- ALREADY_HAVE_PLUS _ -> protoerrhandler proxynextclientmessage $+ ALREADY_HAVE_PLUS _ -> protoerrhandler requestcomplete $ client $ net $ sendMessage resp PUT_FROM _ -> getresponse client resp $ @@ -376,7 +487,7 @@ (const protoerr) _ -> protoerr handlePUT [] _ _ = - protoerrhandler proxynextclientmessage $+ protoerrhandler requestcomplete $ client $ net $ sendMessage ALREADY_HAVE handlePUT remotesides k message = handlePutMulti remotesides k message@@ -388,8 +499,8 @@ relayDATACore len (runRemoteSide remoteside) client $ relayDATAFinish (runRemoteSide remoteside) client $ relayonemessage client (runRemoteSide remoteside) $- const proxynextclientmessage- + const requestcomplete+ relayPUT remoteside k len = relayDATAStart (runRemoteSide remoteside) $ relayDATACore len client (runRemoteSide remoteside) $ relayDATAFinish client (runRemoteSide remoteside) $@@ -397,15 +508,15 @@ where finished resp () = do void $ relayPUTRecord k remoteside resp- proxynextclientmessage ()+ requestcomplete () relayPUTRecord k remoteside SUCCESS = do- addedContent proxymethods (Remote.uuid (remote remoteside)) k+ addedContent (proxyMethods proxyparams) (Remote.uuid (remote remoteside)) k return $ Just [Remote.uuid (remote remoteside)] relayPUTRecord k remoteside (SUCCESS_PLUS us) = do let us' = (Remote.uuid (remote remoteside)) : us forM_ us' $ \u ->- addedContent proxymethods u k+ addedContent (proxyMethods proxyparams) u k return $ Just us' relayPUTRecord _ _ _ = return Nothing@@ -427,14 +538,14 @@ let alreadyhave = \case Right (Left _) -> True _ -> False- l <- forMC concurrencyconfig remotesides initiate+ l <- forMC (proxyConcurrencyConfig proxyparams) remotesides initiate if all alreadyhave l- then if protocolversion < 2- then protoerrhandler proxynextclientmessage $+ then if proxyClientProtocolVersion proxyparams < ProtocolVersion 2+ then protoerrhandler requestcomplete $ client $ net $ sendMessage ALREADY_HAVE- else protoerrhandler proxynextclientmessage $+ else protoerrhandler requestcomplete $ client $ net $ sendMessage $ ALREADY_HAVE_PLUS $- filter (/= remoteuuid) $+ filter (/= proxyUUID proxyparams) $ map (Remote.uuid . remote) (lefts (rights l)) else if null (rights l) -- no response from any remote@@ -447,17 +558,20 @@ (const protoerr) relayPUTMulti minoffset remotes k (Len datalen) _ = do- let totallen = datalen + minoffset -- Tell each remote how much data to expect, depending -- on the remote's offset.- rs <- forMC concurrencyconfig remotes $ \r@(remoteside, remoteoffset) ->+ rs <- forMC (proxyConcurrencyConfig proxyparams) remotes $ \r@(remoteside, remoteoffset) -> runRemoteSideOrSkipFailed remoteside $ do net $ sendMessage $ DATA $ Len $- totallen - remoteoffset+ if remoteoffset > totallen+ then 0+ else totallen - remoteoffset return r protoerrhandler (send (catMaybes rs) minoffset) $ client $ net $ receiveBytes (Len datalen) nullMeterUpdate where+ totallen = datalen + minoffset+ chunksize = fromIntegral defaultChunkSize -- Stream the lazy bytestring out to the remotes in chunks.@@ -467,7 +581,7 @@ let (chunk, b') = L.splitAt chunksize b let chunklen = fromIntegral (L.length chunk) let !n' = n + chunklen- rs' <- forMC concurrencyconfig rs $ \r@(remoteside, remoteoffset) ->+ rs' <- forMC (proxyConcurrencyConfig proxyparams) rs $ \r@(remoteside, remoteoffset) -> if n >= remoteoffset then runRemoteSideOrSkipFailed remoteside $ do net $ sendBytes (Len chunklen) chunk nullMeterUpdate@@ -482,13 +596,21 @@ return r else return (Just r) if L.null b'- then sent (catMaybes rs')+ then do+ -- If we didn't receive as much+ -- data as expected, close+ -- connections to all the remotes,+ -- because they are still waiting+ -- on the rest of the data.+ when (n' /= totallen) $+ mapM_ (closeRemoteSide . fst) rs+ sent (catMaybes rs') else send (catMaybes rs') n' b' sent [] = proxydone sent rs = relayDATAFinishMulti k (map fst rs) - runRemoteSideOrSkipFailed remoteside a = + runRemoteSideOrSkipFailed remoteside a = runRemoteSide remoteside a >>= \case Right v -> return (Just v) Left _ -> do@@ -508,16 +630,16 @@ y $ net $ sendBytes len b nullMeterUpdate relayDATAFinish x y sendsuccessfailure ()- | protocolversion == 0 = sendsuccessfailure+ | proxyClientProtocolVersion proxyparams == ProtocolVersion 0 = sendsuccessfailure -- Protocol version 1 has a VALID or -- INVALID message after the data. | otherwise = relayonemessage x y (\_ () -> sendsuccessfailure) relayDATAFinishMulti k rs- | protocolversion == 0 = + | proxyClientProtocolVersion proxyparams == ProtocolVersion 0 = finish $ net receiveMessage | otherwise =- flip protoerrhandler (client $ net $ receiveMessage) $+ flip protoerrhandler (client $ net receiveMessage) $ withresp $ \message -> finish $ do -- Relay VALID or INVALID message@@ -529,17 +651,17 @@ net receiveMessage where finish a = do- storeduuids <- forMC concurrencyconfig rs $ \r -> + storeduuids <- forMC (proxyConcurrencyConfig proxyparams) rs $ \r -> runRemoteSideOrSkipFailed r a >>= \case Just (Just resp) -> relayPUTRecord k r resp _ -> return Nothing- protoerrhandler proxynextclientmessage $+ protoerrhandler requestcomplete $ client $ net $ sendMessage $ case concat (catMaybes storeduuids) of [] -> FAILURE us- | protocolversion < 2 -> SUCCESS+ | proxyClientProtocolVersion proxyparams < ProtocolVersion 2 -> SUCCESS | otherwise -> SUCCESS_PLUS us -- The associated file received from the P2P protocol@@ -554,10 +676,13 @@ data ConcurrencyConfig = ConcurrencyConfig Int (MSem.MSem Int) noConcurrencyConfig :: Annex ConcurrencyConfig-noConcurrencyConfig = liftIO $ ConcurrencyConfig 1 <$> MSem.new 1+noConcurrencyConfig = mkConcurrencyConfig 1 -getConcurrencyConfig :: Annex ConcurrencyConfig-getConcurrencyConfig = (annexJobs <$> Annex.getGitConfig) >>= \case+mkConcurrencyConfig :: Int -> Annex ConcurrencyConfig+mkConcurrencyConfig n = liftIO $ ConcurrencyConfig n <$> MSem.new n++concurrencyConfigJobs :: Annex ConcurrencyConfig+concurrencyConfigJobs = (annexJobs <$> Annex.getGitConfig) >>= \case NonConcurrent -> noConcurrencyConfig Concurrent n -> go n ConcurrentPerCpu -> go =<< liftIO getNumProcessors@@ -567,8 +692,7 @@ when (n > c) $ liftIO $ setNumCapabilities n setConcurrency (ConcurrencyGitConfig (Concurrent n))- msem <- liftIO $ MSem.new n- return (ConcurrencyConfig n msem)+ mkConcurrencyConfig n forMC :: ConcurrencyConfig -> [a] -> (a -> Annex b) -> Annex [b] forMC _ (x:[]) a = do
Remote/Adb.hs view
@@ -210,7 +210,7 @@ ] remove :: AndroidSerial -> AndroidPath -> Remover-remove serial adir k =+remove serial adir _proof k = unlessM (remove' serial (androidLocation adir k)) $ giveup "adb failed"
Remote/BitTorrent.hs view
@@ -121,8 +121,8 @@ uploadKey :: Key -> AssociatedFile -> Maybe FilePath -> MeterUpdate -> Annex () uploadKey _ _ _ _ = giveup "upload to bittorrent not supported" -dropKey :: Key -> Annex ()-dropKey k = mapM_ (setUrlMissing k) =<< getBitTorrentUrls k+dropKey :: Maybe SafeDropProof -> Key -> Annex ()+dropKey _ k = mapM_ (setUrlMissing k) =<< getBitTorrentUrls k {- We punt and don't try to check if a torrent has enough seeders - with all the pieces etc. That would be quite hard.. and even if
Remote/Bup.hs view
@@ -210,7 +210,7 @@ - We can, however, remove the git branch that bup created for the key. -} remove :: BupRepo -> Remover-remove buprepo k = do+remove buprepo _proof k = do go =<< liftIO (bup2GitRemote buprepo) warning "content cannot be completely removed from bup remote" where
Remote/Ddar.hs view
@@ -180,7 +180,7 @@ go _ _ _ = error "internal" remove :: DdarRepo -> Remover-remove ddarrepo key = do+remove ddarrepo _proof key = do (cmd, params) <- ddarRemoteCall NoConsumeStdin ddarrepo 'd' [Param $ serializeKey key] unlessM (liftIO $ boolSystem cmd params) $
Remote/Directory.hs view
@@ -270,7 +270,7 @@ #endif removeKeyM :: RawFilePath -> Remover-removeKeyM d k = liftIO $ removeDirGeneric True+removeKeyM d _proof k = liftIO $ removeDirGeneric True (fromRawFilePath d) (fromRawFilePath (storeDir d k))
Remote/External.hs view
@@ -259,7 +259,7 @@ _ -> Nothing removeKeyM :: External -> Remover-removeKeyM external k = either giveup return =<< go+removeKeyM external _proof k = either giveup return =<< go where go = handleRequestKey external REMOVE k Nothing $ \resp -> case resp of
Remote/GCrypt.hs view
@@ -432,12 +432,12 @@ retrieversync = fileRetriever $ Remote.Rsync.retrieve rsyncopts remove :: Remote -> Remote.Rsync.RsyncOpts -> AccessMethod -> Remover-remove r rsyncopts accessmethod k = do+remove r rsyncopts accessmethod proof k = do repo <- getRepo r- remove' repo r rsyncopts accessmethod k+ remove' repo r rsyncopts accessmethod proof k remove' :: Git.Repo -> Remote -> Remote.Rsync.RsyncOpts -> AccessMethod -> Remover-remove' repo r rsyncopts accessmethod k+remove' repo r rsyncopts accessmethod proof k | not $ Git.repoIsUrl repo = guardUsable repo (giveup "cannot access remote") $ liftIO $ Remote.Directory.removeDirGeneric True (gCryptTopDir repo)@@ -446,8 +446,8 @@ | accessmethod == AccessRsyncOverSsh = removersync | otherwise = unsupportedUrl where- removersync = Remote.Rsync.remove rsyncopts k- removeshell = Ssh.dropKey repo k+ removersync = Remote.Rsync.remove rsyncopts proof k+ removeshell = Ssh.dropKey repo proof k checkKey :: Remote -> Remote.Rsync.RsyncOpts -> AccessMethod -> CheckPresent checkKey r rsyncopts accessmethod k = do
Remote/Git.hs view
@@ -57,10 +57,14 @@ import qualified Remote.GitLFS import qualified Remote.P2P import qualified Remote.Helper.P2P as P2PHelper+import qualified P2P.Protocol as P2P import P2P.Address+import P2P.Http.Url+import P2P.Http.Client import Annex.Path import Creds import Types.NumCopies+import Annex.SafeDropProof import Types.ProposedAccepted import Annex.Action import Messages.Progress@@ -102,15 +106,21 @@ proxied <- listProxied proxies rs' return (proxied++rs') where- annexurl r = remoteConfig r "annexurl" tweakurl c r = do let n = fromJust $ Git.remoteName r- case M.lookup (annexurl r) c of- Nothing -> return r- Just url -> inRepo $ \g ->- Git.Construct.remoteNamed n $- Git.Construct.fromRemoteLocation (Git.fromConfigValue url) False g+ case getAnnexUrl r c of+ Just url | not (isP2PHttpProtocolUrl url) -> + inRepo $ \g -> Git.Construct.remoteNamed n $+ Git.Construct.fromRemoteLocation url+ False g+ _ -> return r +getAnnexUrl :: Git.Repo -> M.Map Git.ConfigKey Git.ConfigValue -> Maybe String+getAnnexUrl r c = Git.fromConfigValue <$> M.lookup (annexUrlConfigKey r) c++annexUrlConfigKey :: Git.Repo -> Git.ConfigKey+annexUrlConfigKey r = remoteConfig r "annexurl"+ isGitRemoteAnnex :: Git.Repo -> Bool isGitRemoteAnnex r = "annex::" `isPrefixOf` Git.repoLocation r @@ -159,8 +169,9 @@ - done each time git-annex is run in a way that uses remotes, unless - annex-checkuuid is false. -- - Conversely, the config of an URL remote is only read when there is no- - cached UUID value. -}+ - The config of other URL remotes is only read when there is no+ - cached UUID value. + -} configRead :: Bool -> Git.Repo -> Annex Git.Repo configRead autoinit r = do gc <- Annex.getRemoteGitConfig r@@ -176,7 +187,6 @@ Just r' -> return r' _ -> return r - gen :: Git.Repo -> UUID -> RemoteConfig -> RemoteGitConfig -> RemoteStateHandle -> Annex (Maybe Remote) gen r u rc gc rs -- Remote.GitLFS may be used with a repo that is also encrypted@@ -217,7 +227,7 @@ , localpath = localpathCalc r , getRepo = getRepoFromState st , gitconfig = gc- , readonly = Git.repoIsHttp r+ , readonly = Git.repoIsHttp r && not (isP2PHttp' gc) , appendonly = False , untrustworthy = False , availability = repoAvail r@@ -235,7 +245,7 @@ | otherwise = expensiveRemoteCost unavailable :: Git.Repo -> UUID -> RemoteConfig -> RemoteGitConfig -> RemoteStateHandle -> Annex (Maybe Remote)-unavailable r = gen r'+unavailable r u c gc = gen r' u c gc' where r' = case Git.location r of Git.Local { Git.gitdir = d } ->@@ -246,6 +256,10 @@ in r { Git.location = Git.Url (url { uriAuthority = Just auth' })} Nothing -> r { Git.location = Git.Unknown } _ -> r -- already unavailable+ gc' = gc+ { remoteAnnexP2PHttpUrl =+ unavailableP2PHttpUrl <$> remoteAnnexP2PHttpUrl gc+ } {- Tries to read the config for a specified remote, updates state, and - returns the updated repo. -}@@ -307,6 +321,12 @@ -- optimisation. unless (fromMaybe False $ Git.Config.isBare r') $ setremote setRemoteBare False+ -- When annex.url is set to a P2P http url,+ -- store in remote.name.annexUrl+ case Git.fromConfigValue <$> Git.Config.getMaybe (annexConfig "url") r' of+ Just u | isP2PHttpProtocolUrl u ->+ setremote (setConfig . annexUrlConfigKey) u+ _ -> noop return r' Left err -> do set_ignore "not usable by git-annex" False@@ -404,10 +424,12 @@ inAnnex' :: Git.Repo -> Remote -> State -> Key -> Annex Bool inAnnex' repo rmt st@(State connpool duc _ _ _) key+ | isP2PHttp rmt = checkp2phttp | Git.repoIsHttp repo = checkhttp | Git.repoIsUrl repo = checkremote | otherwise = checklocal where+ checkp2phttp = p2pHttpClient rmt giveup (clientCheckPresent key) checkhttp = do gc <- Annex.getGitConfig Url.withUrlOptionsPromptingCreds $ \uo -> @@ -437,34 +459,70 @@ #endif remoteconfig = gitconfig r -dropKey :: Remote -> State -> Key -> Annex ()-dropKey r st key = do+dropKey :: Remote -> State -> Maybe SafeDropProof -> Key -> Annex ()+dropKey r st proof key = do repo <- getRepo r- dropKey' repo r st key+ dropKey' repo r st proof key -dropKey' :: Git.Repo -> Remote -> State -> Key -> Annex ()-dropKey' repo r st@(State connpool duc _ _ _) key+dropKey' :: Git.Repo -> Remote -> State -> Maybe SafeDropProof -> Key -> Annex ()+dropKey' repo r st@(State connpool duc _ _ _) proof key+ | isP2PHttp r = + clientRemoveWithProof proof key unabletoremove r >>= \case+ RemoveResultPlus True fanoutuuids ->+ storefanout fanoutuuids+ RemoveResultPlus False fanoutuuids -> do+ storefanout fanoutuuids+ unabletoremove | not $ Git.repoIsUrl repo = ifM duc- ( guardUsable repo (giveup "cannot access remote") $- commitOnCleanup repo r st $ onLocalFast st $ do- whenM (Annex.Content.inAnnex key) $ do- let cleanup = logStatus key InfoMissing- Annex.Content.lockContentForRemoval key cleanup $ \lock -> do- Annex.Content.removeAnnex lock- cleanup+ ( guardUsable repo (giveup "cannot access remote") removelocal , giveup "remote does not have expected annex.uuid value" )- | Git.repoIsHttp repo = giveup "dropping from http remote not supported"- | otherwise = P2PHelper.remove (uuid r) - (Ssh.runProto r connpool (return (Right False, Nothing))) key+ | Git.repoIsHttp repo = giveup "dropping from this remote is not supported"+ | otherwise = P2PHelper.remove (uuid r) p2prunner proof key+ where+ p2prunner = Ssh.runProto r connpool (return (Right False, Nothing)) + unabletoremove = giveup "removing content from remote failed"++ -- It could take a long time to eg, automount a drive containing+ -- the repo, so check the proof for expiry again after locking the+ -- content for removal.+ removelocal = do+ proofunexpired <- commitOnCleanup repo r st $ onLocalFast st $ do+ ifM (Annex.Content.inAnnex key)+ ( do+ let cleanup = do+ logStatus key InfoMissing+ return True+ Annex.Content.lockContentForRemoval key cleanup $ \lock ->+ ifM (liftIO $ checkSafeDropProofEndTime proof) + ( do+ Annex.Content.removeAnnex lock+ cleanup+ , return False+ )+ , return True+ )+ unless proofunexpired+ safeDropProofExpired+ + storefanout = P2PHelper.storeFanout key InfoMissing (uuid r) . map fromB64UUID+ lockKey :: Remote -> State -> Key -> (VerifiedCopy -> Annex r) -> Annex r-lockKey r st key callback = do+lockKey r st key callback = do repo <- getRepo r lockKey' repo r st key callback lockKey' :: Git.Repo -> Remote -> State -> Key -> (VerifiedCopy -> Annex r) -> Annex r lockKey' repo r st@(State connpool duc _ _ _) key callback+ | isP2PHttp r = do + showLocking r+ p2pHttpClient r giveup (clientLockContent key) >>= \case+ LockResult True (Just lckid) ->+ p2pHttpClient r failedlock $+ clientKeepLocked lckid (uuid r)+ failedlock callback+ _ -> failedlock | not $ Git.repoIsUrl repo = ifM duc ( guardUsable repo failedlock $ do inorigrepo <- Annex.makeRunner@@ -472,7 +530,7 @@ -- and then run the callback in the original -- annex monad, not the remote's. onLocalFast st $ - Annex.Content.lockContentShared key $+ Annex.Content.lockContentShared key Nothing $ liftIO . inorigrepo . callback , failedlock )@@ -491,7 +549,8 @@ copyFromRemote'' repo r st key file dest meterupdate vc copyFromRemote'' :: Git.Repo -> Remote -> State -> Key -> AssociatedFile -> FilePath -> MeterUpdate -> VerifyConfig -> Annex Verification-copyFromRemote'' repo r st@(State connpool _ _ _ _) key file dest meterupdate vc+copyFromRemote'' repo r st@(State connpool _ _ _ _) key af dest meterupdate vc+ | isP2PHttp r = copyp2phttp | Git.repoIsHttp repo = verifyKeyContentIncrementally vc key $ \iv -> do gc <- Annex.getGitConfig ok <- Url.withUrlOptionsPromptingCreds $@@ -501,8 +560,6 @@ | not $ Git.repoIsUrl repo = guardUsable repo (giveup "cannot access remote") $ do u <- getUUID hardlink <- wantHardLink- let bwlimit = remoteAnnexBwLimitDownload (gitconfig r)- <|> remoteAnnexBwLimit (gitconfig r) -- run copy from perspective of remote onLocalFast st $ Annex.Content.prepSendAnnex' key Nothing >>= \case Just (object, _sz, check) -> do@@ -511,7 +568,7 @@ Nothing -> return True copier <- mkFileCopier hardlink st (ok, v) <- runTransfer (Transfer Download u (fromKey id key))- Nothing file Nothing stdRetry $ \p ->+ Nothing af Nothing stdRetry $ \p -> metered (Just (combineMeterUpdate p meterupdate)) key bwlimit $ \_ p' -> copier object dest key p' checksuccess vc if ok@@ -522,8 +579,26 @@ P2PHelper.retrieve (gitconfig r) (Ssh.runProto r connpool (return (False, UnVerified)))- key file dest meterupdate vc- | otherwise = giveup "copying from non-ssh, non-http remote not supported"+ key af dest meterupdate vc+ | otherwise = giveup "copying from this remote is not supported"+ where+ bwlimit = remoteAnnexBwLimitDownload (gitconfig r)+ <|> remoteAnnexBwLimit (gitconfig r)+ + copyp2phttp = verifyKeyContentIncrementally vc key $ \iv -> do+ startsz <- liftIO $ tryWhenExists $+ getFileSize (toRawFilePath dest)+ bracketIO (openBinaryFile dest ReadWriteMode) (hClose) $ \h -> do+ metered (Just meterupdate) key bwlimit $ \_ p -> do+ p' <- case startsz of+ Just startsz' -> liftIO $ do+ resumeVerifyFromOffset startsz' iv p h + _ -> return p+ let consumer = meteredWrite' p' + (writeVerifyChunk iv h)+ p2pHttpClient r giveup (clientGet key af consumer startsz) >>= \case+ Valid -> return ()+ Invalid -> giveup "Transfer failed" copyFromRemoteCheap :: State -> Git.Repo -> Maybe (Key -> AssociatedFile -> FilePath -> Annex ()) #ifndef mingw32_HOST_OS@@ -550,9 +625,10 @@ copyToRemote' :: Git.Repo -> Remote -> State -> Key -> AssociatedFile -> Maybe FilePath -> MeterUpdate -> Annex () copyToRemote' repo r st@(State connpool duc _ _ _) key af o meterupdate+ | isP2PHttp r = prepsendwith copyp2phttp | not $ Git.repoIsUrl repo = ifM duc ( guardUsable repo (giveup "cannot access remote") $ commitOnCleanup repo r st $- copylocal =<< Annex.Content.prepSendAnnex' key o+ prepsendwith copylocal , giveup "remote does not have expected annex.uuid value" ) | Git.repoIsSsh repo =@@ -560,18 +636,24 @@ (Ssh.runProto r connpool (return Nothing)) key af o meterupdate - | otherwise = giveup "copying to non-ssh repo not supported"+ | otherwise = giveup "copying to this remote is not supported" where- copylocal Nothing = giveup "content not available"- copylocal (Just (object, sz, check)) = do+ prepsendwith a = Annex.Content.prepSendAnnex' key o >>= \case+ Nothing -> giveup "content not available"+ Just v -> a v+ + bwlimit = remoteAnnexBwLimitUpload (gitconfig r)+ <|> remoteAnnexBwLimit (gitconfig r)++ failedsend = giveup "failed to send content to remote"++ copylocal (object, sz, check) = do -- The check action is going to be run in -- the remote's Annex, but it needs access to the local -- Annex monad's state. checkio <- Annex.withCurrentState check u <- getUUID hardlink <- wantHardLink- let bwlimit = remoteAnnexBwLimitUpload (gitconfig r)- <|> remoteAnnexBwLimit (gitconfig r) -- run copy from perspective of remote res <- onLocalFast st $ ifM (Annex.Content.inAnnex key) ( return True@@ -587,8 +669,31 @@ copier object (fromRawFilePath dest) key p' checksuccess verify ) unless res $- giveup "failed to send content to remote"+ failedsend + copyp2phttp (object, sz, check) =+ let check' = check >>= \case+ Just s -> do+ warning (UnquotedString s)+ return False+ Nothing -> return True+ in p2pHttpClient r (const $ pure $ PutOffsetResultPlus (Offset 0)) (clientPutOffset key) >>= \case+ PutOffsetResultPlus (offset@(Offset (P2P.Offset n))) ->+ metered (Just meterupdate) key bwlimit $ \_ p -> do+ let p' = offsetMeterUpdate p (BytesProcessed n)+ res <- p2pHttpClient r giveup $+ clientPut p' key (Just offset) af object sz check'+ case res of+ PutResultPlus False fanoutuuids -> do+ storefanout fanoutuuids+ failedsend+ PutResultPlus True fanoutuuids ->+ storefanout fanoutuuids+ PutOffsetResultAlreadyHavePlus fanoutuuids ->+ storefanout fanoutuuids+ + storefanout = P2PHelper.storeFanout key InfoPresent (uuid r) . map fromB64UUID+ fsckOnRemote :: Git.Repo -> [CommandParam] -> Annex (IO Bool) fsckOnRemote r params | Git.repoIsUrl r = return $ return False@@ -847,7 +952,7 @@ adduuid ck = M.insert ck [Git.ConfigValue $ fromUUID $ proxyRemoteUUID p] - addurl = M.insert (remoteConfig renamedr (remoteGitConfigKey UrlField))+ addurl = M.insert (mkRemoteConfigKey renamedr (remoteGitConfigKey UrlField)) [Git.ConfigValue $ encodeBS $ Git.repoLocation r] addproxiedby = case remoteAnnexUUID gc of@@ -875,7 +980,7 @@ proxieduuids = S.map proxyRemoteUUID proxied addremoteannexfield f = M.insert- (remoteAnnexConfig renamedr (remoteGitConfigKey f))+ (mkRemoteConfigKey renamedr (remoteGitConfigKey f)) inheritconfigs c = foldl' inheritconfig c proxyInheritedFields @@ -883,8 +988,8 @@ (Nothing, Just v) -> M.insert dest v c _ -> c where- src = remoteAnnexConfig r k- dest = remoteAnnexConfig renamedr k+ src = mkRemoteConfigKey r k+ dest = mkRemoteConfigKey renamedr k -- When the git config has anything set for a remote, -- avoid making a proxied remote with the same name.@@ -901,7 +1006,15 @@ -- Proxing is also yet supported for remotes using P2P -- addresses. canproxy gc r+ | isP2PHttp' gc = True | remoteAnnexGitLFS gc = False | Git.GCrypt.isEncrypted r = False | Git.repoIsLocal r || Git.repoIsLocalUnknown r = False | otherwise = isNothing (repoP2PAddress r)++isP2PHttp :: Remote -> Bool+isP2PHttp = isP2PHttp' . gitconfig++isP2PHttp' :: RemoteGitConfig -> Bool+isP2PHttp' = isJust . remoteAnnexP2PHttpUrl+
Remote/GitLFS.hs view
@@ -504,7 +504,7 @@ lockKey :: Remote -> RemoteStateHandle -> TVar LFSHandle -> Key -> (VerifiedCopy -> Annex a) -> Annex a lockKey r rs h key callback = ifM (checkKey rs h key)- ( withVerifiedCopy LockedCopy (uuid r) (return True) callback+ ( withVerifiedCopy LockedCopy (uuid r) (return (Right True)) callback , giveup $ "content seems to be missing from " ++ name r )
Remote/Glacier.hs view
@@ -214,7 +214,7 @@ go' _ _ = error "internal" remove :: Remote -> Remover-remove r k = unlessM go $+remove r _proof k = unlessM go $ giveup "removal from glacier failed" where go = glacierAction r
Remote/Helper/Chunked.hs view
@@ -226,10 +226,10 @@ - - This action may be called on a chunked key. It will simply remove it. -}-removeChunks :: Remover -> UUID -> ChunkConfig -> EncKey -> Key -> Annex ()-removeChunks remover u chunkconfig encryptor k = do+removeChunks :: Remover -> UUID -> ChunkConfig -> EncKey -> Maybe SafeDropProof -> Key -> Annex ()+removeChunks remover u chunkconfig encryptor proof k = do ls <- map chunkKeyList <$> chunkKeys u chunkconfig k- mapM_ (remover . encryptor) (concat ls)+ mapM_ (remover proof . encryptor) (concat ls) let chunksizes = catMaybes $ map (fromKey keyChunkSize <=< headMaybe) ls forM_ chunksizes $ chunksRemoved u k . FixedSizeChunks . fromIntegral
Remote/Helper/Hooks.hs view
@@ -41,7 +41,8 @@ , retrieveKeyFileCheap = case retrieveKeyFileCheap r of Just a -> Just $ \k af f -> wrapper $ a k af f Nothing -> Nothing- , removeKey = wrapper . removeKey r+ , removeKey = \proof k ->+ wrapper $ removeKey r proof k , checkPresent = wrapper . checkPresent r } where
Remote/Helper/P2P.hs view
@@ -21,8 +21,10 @@ import Annex.Verify import Logs.Location import Utility.SafeOutput+import Utility.HumanTime import Control.Concurrent+import Data.Time.Clock.POSIX -- Runs a Proto action using a connection it sets up. type ProtoRunner a = P2P.Proto a -> Annex (Maybe a)@@ -40,16 +42,17 @@ let bwlimit = remoteAnnexBwLimitUpload gc <|> remoteAnnexBwLimit gc metered (Just p) sizer bwlimit $ \_ p' -> runner (P2P.put k af p') >>= \case- Just (Just fanoutuuids) -> do- -- Storing on the remote can cause it- -- to be stored on additional UUIDs, - -- so record those.- forM_ fanoutuuids $ \u ->- when (u /= remoteuuid) $- logChange k u InfoPresent+ Just (Just fanoutuuids) -> + storeFanout k InfoPresent remoteuuid fanoutuuids Just Nothing -> giveup "Transfer failed" Nothing -> remoteUnavail +storeFanout :: Key -> LogStatus -> UUID -> [UUID] -> Annex ()+storeFanout k logstatus remoteuuid us = + forM_ us $ \u ->+ when (u /= remoteuuid) $+ logChange k u logstatus+ retrieve :: RemoteGitConfig -> (ProtoRunner (Bool, Verification)) -> Key -> AssociatedFile -> FilePath -> MeterUpdate -> VerifyConfig -> Annex Verification retrieve gc runner k af dest p verifyconfig = do iv <- startVerifyKeyContentIncrementally verifyconfig k@@ -60,22 +63,18 @@ Just (False, _) -> giveup "Transfer failed" Nothing -> remoteUnavail -remove :: UUID -> ProtoRunner (Either String Bool, Maybe [UUID]) -> Key -> Annex ()-remove remoteuuid runner k = runner (P2P.remove k) >>= \case- Just (Right True, alsoremoveduuids) -> note alsoremoveduuids+remove :: UUID -> ProtoRunner (Either String Bool, Maybe [UUID]) -> Maybe SafeDropProof -> Key -> Annex ()+remove remoteuuid runner proof k = runner (P2P.remove proof k) >>= \case+ Just (Right True, alsoremoveduuids) -> + storeFanout k InfoMissing remoteuuid+ (fromMaybe [] alsoremoveduuids) Just (Right False, alsoremoveduuids) -> do- note alsoremoveduuids+ storeFanout k InfoMissing remoteuuid+ (fromMaybe [] alsoremoveduuids) giveup "removing content from remote failed" Just (Left err, _) -> do giveup (safeOutput err) Nothing -> remoteUnavail- where- -- The remote reports removal from other UUIDs than its own,- -- so record those.- note alsoremoveduuids = - forM_ (fromMaybe [] alsoremoveduuids) $ \u ->- when (u /= remoteuuid) $- logChange k u InfoMissing checkpresent :: ProtoRunner (Either String Bool) -> Key -> Annex Bool checkpresent runner k =@@ -85,20 +84,32 @@ Just (Right b) -> return b Just (Left err) -> giveup (safeOutput err) +{- Locks the content on the remote while running an action with a+ - LockedCopy.+ -+ - Note that this only guarantees that the content is locked as long as the+ - connection to the peer remains up. If the connection is unexpectededly+ - dropped, the peer will then unlock the content.+ -} lock :: WithConn a c -> ProtoConnRunner c -> UUID -> Key -> (VerifiedCopy -> Annex a) -> Annex a lock withconn connrunner u k callback = withconn $ \conn -> do+ starttime <- liftIO getPOSIXTime connv <- liftIO $ newMVar conn let runproto d p = do c <- liftIO $ takeMVar connv (c', mr) <- connrunner p c liftIO $ putMVar connv c' return (fromMaybe d mr)- r <- P2P.lockContentWhile runproto k go+ r <- P2P.lockContentWhile runproto k (go starttime) conn' <- liftIO $ takeMVar connv return (conn', r) where- go False = giveup "can't lock content"- go True = withVerifiedCopy LockedCopy u (return True) callback+ go _ False = giveup "can't lock content"+ go starttime True = do+ let check = return $ Left $ starttime + retentionduration+ withVerifiedCopy LockedCopy u check callback+ retentionduration = fromIntegral $+ durationSeconds p2pDefaultLockContentRetentionDuration remoteUnavail :: a remoteUnavail = giveup "can't connect to remote"
Remote/Helper/ReadOnly.hs view
@@ -47,8 +47,8 @@ readonlyStoreKey :: Key -> AssociatedFile -> Maybe FilePath -> MeterUpdate -> Annex () readonlyStoreKey _ _ _ _ = readonlyFail -readonlyRemoveKey :: Key -> Annex ()-readonlyRemoveKey _ = readonlyFail+readonlyRemoveKey :: Maybe SafeDropProof -> Key -> Annex ()+readonlyRemoveKey _ _ = readonlyFail readonlyStorer :: Storer readonlyStorer _ _ _ = readonlyFail
Remote/Helper/Special.hs view
@@ -138,8 +138,8 @@ storeKeyDummy _ _ _ _ = error "missing storeKey implementation" retrieveKeyFileDummy :: Key -> AssociatedFile -> FilePath -> MeterUpdate -> VerifyConfig -> Annex Verification retrieveKeyFileDummy _ _ _ _ _ = error "missing retrieveKeyFile implementation"-removeKeyDummy :: Key -> Annex ()-removeKeyDummy _ = error "missing removeKey implementation"+removeKeyDummy :: Maybe SafeDropProof -> Key -> Annex ()+removeKeyDummy _ _ = error "missing removeKey implementation" checkPresentDummy :: Key -> Annex Bool checkPresentDummy _ = error "missing checkPresent implementation" @@ -197,7 +197,7 @@ , retrievalSecurityPolicy = if isencrypted then mkRetrievalVerifiableKeysSecure (gitconfig baser) else retrievalSecurityPolicy baser- , removeKey = \k -> cip >>= removeKeyGen k+ , removeKey = \k proof -> cip >>= removeKeyGen k proof , checkPresent = \k -> cip >>= checkPresentGen k , cost = if isencrypted then cost baser + encryptedRemoteCostAdj@@ -227,7 +227,7 @@ storeChunks (uuid baser) chunkconfig enck k src p' enc encr storer checkpresent where- rollback = void $ removeKey encr k+ rollback = void $ removeKey encr Nothing k enck = maybe id snd enc -- call retriever to get chunks; decrypt them; stream to dest file@@ -238,8 +238,8 @@ where enck = maybe id snd enc - removeKeyGen k enc = - removeChunks remover (uuid baser) chunkconfig enck k+ removeKeyGen proof k enc = + removeChunks remover (uuid baser) chunkconfig enck proof k where enck = maybe id snd enc
Remote/Helper/Ssh.hs view
@@ -44,12 +44,13 @@ git_annex_shell :: ConsumeStdin -> Git.Repo -> String -> [CommandParam] -> [(Field, String)] -> Annex (Maybe (FilePath, [CommandParam])) git_annex_shell cs r command params fields | not $ Git.repoIsUrl r = do- shellopts <- getshellopts+ dir <- liftIO $ absPath (Git.repoPath r)+ shellopts <- getshellopts dir return $ Just (shellcmd, shellopts ++ fieldopts) | Git.repoIsSsh r = do gc <- Annex.getRemoteGitConfig r u <- getRepoUUID r- shellopts <- getshellopts+ shellopts <- getshellopts (Git.repoPath r) let sshcmd = unwords $ fromMaybe shellcmd (remoteAnnexShell gc) : map shellEscape (toCommand shellopts) ++@@ -58,9 +59,8 @@ Just <$> toRepo cs r gc sshcmd | otherwise = return Nothing where- dir = Git.repoPath r shellcmd = "git-annex-shell"- getshellopts = do+ getshellopts dir = do debugenabled <- Annex.getRead Annex.debugenabled debugselector <- Annex.getRead Annex.debugselector let params' = case (debugenabled, debugselector) of@@ -105,13 +105,17 @@ dispatch (ExitFailure 1) = return False dispatch _ = cantCheck r -{- Removes a key from a remote. -}-dropKey :: Git.Repo -> Key -> Annex ()-dropKey r key = unlessM (dropKey' r key) $+{- Removes a key from a remote using the legacy git-annex-shell dropkey,+ - rather than the P2P protocol.+ -+ - The proof is not checked for expire on the remote, so this should only+ - be used by remotes that do not have lockContent return a LockedCopy. -}+dropKey :: Git.Repo -> Maybe SafeDropProof -> Key -> Annex ()+dropKey r proof key = unlessM (dropKey' r proof key) $ giveup "unable to remove key from remote" -dropKey' :: Git.Repo -> Key -> Annex Bool-dropKey' r key = onRemote NoConsumeStdin r (\f p -> liftIO (boolSystem f p), return False) "dropkey"+dropKey' :: Git.Repo -> Maybe SafeDropProof -> Key -> Annex Bool+dropKey' r _proof key = onRemote NoConsumeStdin r (\f p -> liftIO (boolSystem f p), return False) "dropkey" [ Param "--quiet", Param "--force" , Param $ serializeKey key ]
Remote/Hook.hs view
@@ -166,7 +166,7 @@ giveup "failed to retrieve content" remove :: HookName -> Remover-remove h k = +remove h _proof k = unlessM (runHook' h "remove" k Nothing $ return True) $ giveup "failed to remove content"
Remote/Rsync.hs view
@@ -265,7 +265,7 @@ ) remove :: RsyncOpts -> Remover-remove o k = removeGeneric o includes+remove o _proof k = removeGeneric o includes where includes = concatMap use dirHashes use h = let dir = fromRawFilePath (h def k) in
Remote/S3.hs view
@@ -449,7 +449,7 @@ Url.sinkResponseFile p iv zeroBytesProcessed f WriteMode rsp remove :: S3HandleVar -> Remote -> S3Info -> Remover-remove hv r info k = withS3HandleOrFail (uuid r) hv $ \h -> do+remove hv r info _proof k = withS3HandleOrFail (uuid r) hv $ \h -> do S3.DeleteObjectResponse <- liftIO $ runResourceT $ sendS3Handle h $ S3.DeleteObject (T.pack $ bucketObject info k) (bucket info) return ()@@ -462,7 +462,7 @@ | versioning info = Just $ \k callback -> do checkVersioning info rs k ifM (checkKey hv r rs c info k)- ( withVerifiedCopy LockedCopy (uuid r) (return True) callback+ ( withVerifiedCopy LockedCopy (uuid r) (return (Right True)) callback , giveup $ "content seems to be missing from " ++ name r ++ " despite S3 versioning being enabled" ) | otherwise = Nothing
Remote/Tahoe.hs view
@@ -155,8 +155,8 @@ go (Just cap) = unlessM (liftIO $ requestTahoe hdl "get" [Param cap, File d]) $ giveup "tahoe failed to reteieve content" -remove :: Key -> Annex ()-remove _k = giveup "content cannot be removed from tahoe remote"+remove :: Maybe SafeDropProof -> Key -> Annex ()+remove _ _ = giveup "content cannot be removed from tahoe remote" -- Since content cannot be removed from tahoe (by git-annex), -- nothing needs to be done to lock content there, except for checking that@@ -164,7 +164,7 @@ lockKey :: UUID -> RemoteStateHandle -> TahoeHandle -> Key -> (VerifiedCopy -> Annex a) -> Annex a lockKey u rs hrl k callback = ifM (checkKey rs hrl k)- ( withVerifiedCopy LockedCopy u (return True) callback+ ( withVerifiedCopy LockedCopy u (return (Right True)) callback , giveup $ "content seems to be missing from tahoe remote" )
Remote/Web.hs view
@@ -184,8 +184,8 @@ uploadKey :: Key -> AssociatedFile -> Maybe FilePath -> MeterUpdate -> Annex () uploadKey _ _ _ _ = giveup "upload to web not supported" -dropKey :: UrlIncludeExclude -> Key -> Annex ()-dropKey urlincludeexclude k = mapM_ (setUrlMissing k) =<< getWebUrls' urlincludeexclude k+dropKey :: UrlIncludeExclude -> Maybe SafeDropProof -> Key -> Annex ()+dropKey urlincludeexclude _proof k = mapM_ (setUrlMissing k) =<< getWebUrls' urlincludeexclude k checkKey :: UrlIncludeExclude -> Key -> Annex Bool checkKey urlincludeexclude key = do
Remote/WebDAV.hs view
@@ -185,7 +185,7 @@ withContentM $ httpBodyRetriever d p iv remove :: DavHandleVar -> Remover-remove hv k = withDavHandle hv $ \dav -> liftIO $ goDAV dav $+remove hv _proof k = withDavHandle hv $ \dav -> liftIO $ goDAV dav $ -- Delete the key's whole directory, including any -- legacy chunked files, etc, in a single action. removeHelper (keyDir k)
Types/GitConfig.hs view
@@ -25,8 +25,12 @@ RemoteGitConfigField(..), remoteGitConfigKey, proxyInheritedFields,+ MkRemoteConfigKey,+ mkRemoteConfigKey, ) where +import Debug.Trace+ import Common import qualified Git import qualified Git.Config@@ -55,6 +59,7 @@ import Utility.ThreadScheduler (Seconds(..)) import Utility.Url (Scheme, mkScheme) import Network.Socket (PortNumber)+import P2P.Http.Url import Control.Concurrent.STM import qualified Data.Set as S@@ -104,6 +109,7 @@ , annexSyncMigrations :: Bool , annexDebug :: Bool , annexDebugFilter :: Maybe String+ , annexUrl :: Maybe String , annexWebOptions :: [String] , annexYoutubeDlOptions :: [String] , annexYoutubeDlCommand :: Maybe String@@ -199,6 +205,7 @@ , annexSyncMigrations = getbool (annexConfig "syncmigrations") True , annexDebug = getbool (annexConfig "debug") False , annexDebugFilter = getmaybe (annexConfig "debugfilter")+ , annexUrl = getmaybe (annexConfig "url") , annexWebOptions = getwords (annexConfig "web-options") , annexYoutubeDlOptions = getwords (annexConfig "youtube-dl-options") , annexYoutubeDlCommand = getmaybe (annexConfig "youtube-dl-command")@@ -395,6 +402,7 @@ , remoteAnnexClusterNode :: Maybe [RemoteName] , remoteAnnexClusterGateway :: [ClusterUUID] , remoteUrl :: Maybe String+ , remoteAnnexP2PHttpUrl :: Maybe P2PHttpUrl {- These settings are specific to particular types of remotes - including special remotes. -}@@ -487,12 +495,17 @@ , remoteAnnexClusterGateway = fromMaybe [] $ (mapMaybe (mkClusterUUID . toUUID) . words) <$> getmaybe ClusterGatewayField- , remoteUrl = - case Git.Config.getMaybe (remoteConfig remotename (remoteGitConfigKey UrlField)) r of+ , remoteUrl = traceShow (mkRemoteConfigKey remotename (remoteGitConfigKey UrlField)) $+ case Git.Config.getMaybe (mkRemoteConfigKey remotename (remoteGitConfigKey UrlField)) r of Just (ConfigValue b) | B.null b -> Nothing | otherwise -> Just (decodeBS b) _ -> Nothing+ , remoteAnnexP2PHttpUrl =+ case Git.Config.getMaybe (mkRemoteConfigKey remotename (remoteGitConfigKey AnnexUrlField)) r of+ Just (ConfigValue b) ->+ parseP2PHttpUrl (decodeBS b)+ _ -> Nothing , remoteAnnexShell = getmaybe ShellField , remoteAnnexSshOptions = getoptions SshOptionsField , remoteAnnexRsyncOptions = getoptions RsyncOptionsField@@ -527,8 +540,8 @@ getmaybe' :: RemoteGitConfigField -> Maybe ConfigValue getmaybe' f = let k = remoteGitConfigKey f- in Git.Config.getMaybe (remoteAnnexConfig remotename k) r- <|> Git.Config.getMaybe (annexConfig k) r+ in Git.Config.getMaybe (mkRemoteConfigKey remotename k) r+ <|> Git.Config.getMaybe (mkAnnexConfigKey k) r getoptions k = fromMaybe [] $ words <$> getmaybe k data RemoteGitConfigField@@ -569,6 +582,7 @@ | ClusterNodeField | ClusterGatewayField | UrlField+ | AnnexUrlField | ShellField | SshOptionsField | RsyncOptionsField@@ -594,86 +608,89 @@ | ExternalTypeField deriving (Enum, Bounded) -remoteGitConfigField :: RemoteGitConfigField -> (UnqualifiedConfigKey, ProxyInherited)+remoteGitConfigField :: RemoteGitConfigField -> (MkRemoteConfigKey, ProxyInherited) remoteGitConfigField = \case -- Hard to know the true cost of accessing eg a slow special -- remote via the proxy. The cost of the proxy is the best guess -- so do inherit it.- CostField -> inherited "cost"- CostCommandField -> inherited "cost-command"- IgnoreField -> inherited "ignore"- IgnoreCommandField -> inherited "ignore-command"- SyncField -> inherited "sync"- SyncCommandField -> inherited "sync-command"- PullField -> inherited "pull"- PushField -> inherited "push"- ReadOnlyField -> inherited "readonly"- CheckUUIDField -> uninherited "checkuuid"- VerifyField -> inherited "verify"- TrackingBranchField -> uninherited "tracking-branch"- ExportTrackingField -> uninherited "export-tracking"- TrustLevelField -> uninherited "trustlevel"- StartCommandField -> uninherited "start-command"- StopCommandField -> uninherited "stop-command"- SpeculatePresentField -> inherited "speculate-present"- BareField -> inherited "bare"- RetryField -> inherited "retry"- ForwardRetryField -> inherited "forward-retry"- RetryDelayField -> inherited "retrydelay"- StallDetectionField -> inherited "stalldetection"- StallDetectionUploadField -> inherited "stalldetection-upload"- StallDetectionDownloadField -> inherited "stalldetection-download"- BWLimitField -> inherited "bwlimit"- BWLimitUploadField -> inherited "bwlimit-upload"- BWLimitDownloadField -> inherited "bwlimit-upload"- UUIDField -> uninherited "uuid"- ConfigUUIDField -> uninherited "config-uuid"- SecurityAllowUnverifiedDownloadsField -> inherited "security-allow-unverified-downloads"- MaxGitBundlesField -> inherited "max-git-bundles"- AllowEncryptedGitRepoField -> inherited "allow-encrypted-gitrepo"+ CostField -> inherited True "cost"+ CostCommandField -> inherited True "cost-command"+ IgnoreField -> inherited True "ignore"+ IgnoreCommandField -> inherited True "ignore-command"+ SyncField -> inherited True "sync"+ SyncCommandField -> inherited True "sync-command"+ PullField -> inherited True "pull"+ PushField -> inherited True "push"+ ReadOnlyField -> inherited True "readonly"+ CheckUUIDField -> uninherited True "checkuuid"+ VerifyField -> inherited True "verify"+ TrackingBranchField -> uninherited True "tracking-branch"+ ExportTrackingField -> uninherited True "export-tracking"+ TrustLevelField -> uninherited True "trustlevel"+ StartCommandField -> uninherited True "start-command"+ StopCommandField -> uninherited True "stop-command"+ SpeculatePresentField -> inherited True "speculate-present"+ BareField -> inherited True "bare"+ RetryField -> inherited True "retry"+ ForwardRetryField -> inherited True "forward-retry"+ RetryDelayField -> inherited True "retrydelay"+ StallDetectionField -> inherited True "stalldetection"+ StallDetectionUploadField -> inherited True "stalldetection-upload"+ StallDetectionDownloadField -> inherited True "stalldetection-download"+ BWLimitField -> inherited True "bwlimit"+ BWLimitUploadField -> inherited True "bwlimit-upload"+ BWLimitDownloadField -> inherited True "bwlimit-upload"+ UUIDField -> uninherited True "uuid"+ ConfigUUIDField -> uninherited True "config-uuid"+ SecurityAllowUnverifiedDownloadsField -> inherited True "security-allow-unverified-downloads"+ MaxGitBundlesField -> inherited True "max-git-bundles"+ AllowEncryptedGitRepoField -> inherited True "allow-encrypted-gitrepo" -- Allow proxy chains.- ProxyField -> inherited "proxy"- ProxiedByField -> uninherited "proxied-by"- ClusterNodeField -> uninherited "cluster-node"- ClusterGatewayField -> uninherited "cluster-gateway"- UrlField -> uninherited "url"- ShellField -> inherited "shell"- SshOptionsField -> inherited "ssh-options"- RsyncOptionsField -> inherited "rsync-options"- RsyncDownloadOptionsField -> inherited "rsync-download-options"- RsyncUploadOptionsField -> inherited "rsync-upload-options"- RsyncTransportField -> inherited "rsync-transport"- GnupgOptionsField -> inherited "gnupg-options"- GnupgDecryptOptionsField -> inherited "gnupg-decrypt-options"- SharedSOPCommandField -> inherited "shared-sop-command"- SharedSOPProfileField -> inherited "shared-sop-profile"- RsyncUrlField -> uninherited "rsyncurl"- BupRepoField -> uninherited "buprepo"- BorgRepoField -> uninherited "borgrepo"- TahoeField -> uninherited "tahoe"- BupSplitOptionsField -> uninherited "bup-split-options"- DirectoryField -> uninherited "directory"- AndroidDirectoryField -> uninherited "androiddirectory"- AndroidSerialField -> uninherited "androidserial"- GCryptField -> uninherited "gcrypt"- GitLFSField -> uninherited "git-lfs"- DdarRepoField -> uninherited "ddarrepo"- HookTypeField -> uninherited "hooktype"- ExternalTypeField -> uninherited "externaltype"+ ProxyField -> inherited True "proxy"+ ProxiedByField -> uninherited True "proxied-by"+ ClusterNodeField -> uninherited True "cluster-node"+ ClusterGatewayField -> uninherited True "cluster-gateway"+ UrlField -> uninherited False "url"+ AnnexUrlField -> inherited False "annexurl"+ ShellField -> inherited True "shell"+ SshOptionsField -> inherited True "ssh-options"+ RsyncOptionsField -> inherited True "rsync-options"+ RsyncDownloadOptionsField -> inherited True "rsync-download-options"+ RsyncUploadOptionsField -> inherited True "rsync-upload-options"+ RsyncTransportField -> inherited True "rsync-transport"+ GnupgOptionsField -> inherited True "gnupg-options"+ GnupgDecryptOptionsField -> inherited True "gnupg-decrypt-options"+ SharedSOPCommandField -> inherited True "shared-sop-command"+ SharedSOPProfileField -> inherited True "shared-sop-profile"+ RsyncUrlField -> uninherited True "rsyncurl"+ BupRepoField -> uninherited True "buprepo"+ BorgRepoField -> uninherited True "borgrepo"+ TahoeField -> uninherited True "tahoe"+ BupSplitOptionsField -> uninherited True "bup-split-options"+ DirectoryField -> uninherited True "directory"+ AndroidDirectoryField -> uninherited True "androiddirectory"+ AndroidSerialField -> uninherited True "androidserial"+ GCryptField -> uninherited True "gcrypt"+ GitLFSField -> uninherited True "git-lfs"+ DdarRepoField -> uninherited True "ddarrepo"+ HookTypeField -> uninherited True "hooktype"+ ExternalTypeField -> uninherited True "externaltype" where- inherited f = (f, ProxyInherited True)- uninherited f = (f, ProxyInherited False)+ inherited True f = (MkRemoteAnnexConfigKey f, ProxyInherited True)+ inherited False f = (MkRemoteConfigKey f, ProxyInherited True)+ uninherited True f = (MkRemoteAnnexConfigKey f, ProxyInherited False)+ uninherited False f = (MkRemoteConfigKey f, ProxyInherited False) newtype ProxyInherited = ProxyInherited Bool -- All remote config fields that are inherited from a proxy.-proxyInheritedFields :: [UnqualifiedConfigKey]+proxyInheritedFields :: [MkRemoteConfigKey] proxyInheritedFields = map fst $ filter (\(_, ProxyInherited p) -> p) $ map remoteGitConfigField [minBound..maxBound] -remoteGitConfigKey :: RemoteGitConfigField -> UnqualifiedConfigKey+remoteGitConfigKey :: RemoteGitConfigField -> MkRemoteConfigKey remoteGitConfigKey = fst . remoteGitConfigField notempty :: Maybe String -> Maybe String @@ -685,13 +702,23 @@ dummyRemoteGitConfig = atomically $ extractRemoteGitConfig Git.Construct.fromUnknown "dummy" -type UnqualifiedConfigKey = B.ByteString+data MkRemoteConfigKey+ = MkRemoteAnnexConfigKey B.ByteString+ | MkRemoteConfigKey B.ByteString +mkRemoteConfigKey :: RemoteNameable r => r -> MkRemoteConfigKey -> ConfigKey+mkRemoteConfigKey r (MkRemoteAnnexConfigKey b) = remoteAnnexConfig r b+mkRemoteConfigKey r (MkRemoteConfigKey b) = remoteConfig r b++mkAnnexConfigKey :: MkRemoteConfigKey -> ConfigKey+mkAnnexConfigKey (MkRemoteAnnexConfigKey b) = annexConfig b+mkAnnexConfigKey (MkRemoteConfigKey b) = annexConfig b+ annexConfigPrefix :: B.ByteString annexConfigPrefix = "annex." {- A global annex setting in git config. -}-annexConfig :: UnqualifiedConfigKey -> ConfigKey+annexConfig :: B.ByteString -> ConfigKey annexConfig key = ConfigKey (annexConfigPrefix <> key) class RemoteNameable r where@@ -704,13 +731,13 @@ getRemoteName = id {- A per-remote annex setting in git config. -}-remoteAnnexConfig :: RemoteNameable r => r -> UnqualifiedConfigKey -> ConfigKey+remoteAnnexConfig :: RemoteNameable r => r -> B.ByteString -> ConfigKey remoteAnnexConfig r = remoteConfig r . remoteAnnexConfigEnd -remoteAnnexConfigEnd :: UnqualifiedConfigKey -> UnqualifiedConfigKey+remoteAnnexConfigEnd :: B.ByteString -> B.ByteString remoteAnnexConfigEnd key = "annex-" <> key {- A per-remote setting in git config. -}-remoteConfig :: RemoteNameable r => r -> UnqualifiedConfigKey -> ConfigKey+remoteConfig :: RemoteNameable r => r -> B.ByteString -> ConfigKey remoteConfig r key = ConfigKey $ "remote." <> encodeBS (getRemoteName r) <> "." <> key
Types/NumCopies.hs view
@@ -1,6 +1,6 @@ {- git-annex numcopies types -- - Copyright 2014-2022 Joey Hess <id@joeyh.name>+ - Copyright 2014-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -22,20 +22,24 @@ withVerifiedCopy, isSafeDrop, SafeDropProof,+ safeDropProofEndTime, mkSafeDropProof, ContentRemovalLock(..),+ p2pDefaultLockContentRetentionDuration, ) where import Types.UUID import Types.Key import Utility.Exception (bracketIO) import Utility.Monad+import Utility.HumanTime import qualified Data.Map as M+import Data.Either import Control.Concurrent.MVar import Control.Monad.Catch (MonadMask) import Control.Monad.IO.Class (MonadIO)-import Control.Monad+import Data.Time.Clock.POSIX (POSIXTime) newtype NumCopies = NumCopies Int deriving (Ord, Eq, Show)@@ -77,14 +81,16 @@ | TrustedCopy V {- The strongest proof of the existence of a copy. - Until its associated action is called to unlock it,+ - or connection with a remote repository is lost, - the copy is locked in the repository and is guaranteed- - not to be removed by any git-annex process. -}+ - not to be removed by any git-annex process. Use+ - checkVerifiedCopy to detect loss of connection. -} | LockedCopy V deriving (Show) data V = V { _getUUID :: UUID- , _checkVerifiedCopy :: IO Bool+ , _checkVerifiedCopy :: IO (Either POSIXTime Bool) , _invalidateVerifiedCopy :: IO () } @@ -99,8 +105,19 @@ toV (RecentlyVerifiedCopy v) = v toV (LockedCopy v) = v --- Checks that it's still valid.-checkVerifiedCopy :: VerifiedCopy -> IO Bool+-- Checks that the VerifiedCopy is still valid.+--+-- Invalidation of the VerifiedCopy will make this return False.+-- +-- When the key is being kept locked by a connection to a remote+-- repository, a detected loss of connection will make this+-- return False. +--+-- When the connection could possibly break without being detected+-- immediately, this will return a POSIXTime that is how long the+-- content is guaranteed to remain locked on the remote even if the+-- connection has broken.+checkVerifiedCopy :: VerifiedCopy -> IO (Either POSIXTime Bool) checkVerifiedCopy = _checkVerifiedCopy . toV invalidateVerifiedCopy :: VerifiedCopy -> IO ()@@ -119,15 +136,18 @@ M.fromListWith strongestVerifiedCopy (zip (map toUUID l) l) mkVerifiedCopy :: ToUUID u => (V -> VerifiedCopy) -> u -> VerifiedCopy-mkVerifiedCopy mk u = mk $ V (toUUID u) (return True) (return ())+mkVerifiedCopy mk u = mk $ V (toUUID u) (return (Right True)) (return ()) -invalidatableVerifiedCopy :: ToUUID u => (V -> VerifiedCopy) -> u -> IO Bool -> IO VerifiedCopy+invalidatableVerifiedCopy :: ToUUID u => (V -> VerifiedCopy) -> u -> IO (Either POSIXTime Bool) -> IO VerifiedCopy invalidatableVerifiedCopy mk u check = do v <- newEmptyMVar let invalidate = do _ <- tryPutMVar v () return ()- let check' = isEmptyMVar v <&&> check+ let check' = ifM (isEmptyMVar v)+ ( check+ , pure (Right False)+ ) return $ mk $ V (toUUID u) check' invalidate -- Constructs a VerifiedCopy, and runs the action, ensuring that the@@ -136,7 +156,7 @@ :: (MonadMask m, MonadIO m, ToUUID u) => (V -> VerifiedCopy) -> u- -> IO Bool+ -> IO (Either POSIXTime Bool) -> (VerifiedCopy -> m a) -> m a withVerifiedCopy mk u check = bracketIO setup cleanup@@ -155,13 +175,26 @@ - to fall below NumCopies, but it will never fall below MinCopies. -} isSafeDrop :: NumCopies -> MinCopies -> [VerifiedCopy] -> Maybe ContentRemovalLock -> Bool+isSafeDrop n m l lck = case safeDropAnalysis n m l lck of+ UnsafeDrop -> False+ SafeDrop -> True+ SafeDropCheckTime -> True++data SafeDropAnalysis+ = UnsafeDrop+ | SafeDrop+ | SafeDropCheckTime++safeDropAnalysis :: NumCopies -> MinCopies -> [VerifiedCopy] -> Maybe ContentRemovalLock -> SafeDropAnalysis {- When a ContentRemovalLock is provided, the content is being - dropped from the local repo. That lock will prevent other git repos - that are concurrently dropping from using the local copy as a VerifiedCopy. - So, no additional locking is needed; all we need is verifications - of any kind of enough other copies of the content. -}-isSafeDrop (NumCopies n) (MinCopies m) l (Just (ContentRemovalLock _)) = - length (deDupVerifiedCopies l) >= max n m+safeDropAnalysis (NumCopies n) (MinCopies m) l (Just (ContentRemovalLock _)) =+ if length (deDupVerifiedCopies l) >= max n m+ then SafeDrop+ else UnsafeDrop {- Dropping from a remote repo. - - To guarantee MinCopies is never violated, at least that many LockedCopy@@ -174,27 +207,49 @@ - violated, this is the best that can be done without requiring that - all special remotes support locking. -}-isSafeDrop (NumCopies n) (MinCopies m) l Nothing- | n == 0 && m == 0 = True- | otherwise = and- [ length (deDupVerifiedCopies l) >= n- , length (filter fullVerification l) >= m- ]+safeDropAnalysis (NumCopies n) (MinCopies m) l Nothing+ | n == 0 && m == 0 = SafeDrop+ | length (deDupVerifiedCopies l) >= n + && length (filter fullVerification l) >= m =+ SafeDropCheckTime+ | otherwise = UnsafeDrop fullVerification :: VerifiedCopy -> Bool fullVerification (LockedCopy _) = True fullVerification (TrustedCopy _) = True fullVerification (RecentlyVerifiedCopy _) = False --- A proof that it's currently safe to drop an object.-data SafeDropProof = SafeDropProof NumCopies MinCopies [VerifiedCopy] (Maybe ContentRemovalLock)+-- Content locked using the P2P protocol defaults to being retained,+-- still locked, for 10 minutes after a connection loss.+-- +-- This is only the case since git-annex 10.20240704, but currently+-- this is used even for older remotes, to avoid a disruptive behavior+-- change when used with remotes running an old version of git-annex.+p2pDefaultLockContentRetentionDuration :: Duration+p2pDefaultLockContentRetentionDuration = Duration (60*10)++-- A proof that it's safe to drop an object.+--+-- It may only be safe up until a given POSIXTime.+data SafeDropProof = SafeDropProof NumCopies MinCopies [VerifiedCopy] (Maybe POSIXTime) (Maybe ContentRemovalLock) deriving (Show) +safeDropProofEndTime :: SafeDropProof -> Maybe POSIXTime+safeDropProofEndTime (SafeDropProof _ _ _ t _) = t+ -- Makes sure that none of the VerifiedCopies have become invalidated--- before constructing proof.+-- before constructing the proof. mkSafeDropProof :: NumCopies -> MinCopies -> [VerifiedCopy] -> Maybe ContentRemovalLock -> IO (Either [VerifiedCopy] SafeDropProof) mkSafeDropProof need mincopies have removallock = do- stillhave <- filterM checkVerifiedCopy have- return $ if isSafeDrop need mincopies stillhave removallock- then Right (SafeDropProof need mincopies stillhave removallock)- else Left stillhave+ l <- mapM checkVerifiedCopy have+ let stillhave = map fst $ + filter (either (const True) id . snd) (zip have l)+ return $ case safeDropAnalysis need mincopies stillhave removallock of+ SafeDrop -> Right $+ SafeDropProof need mincopies stillhave Nothing removallock+ SafeDropCheckTime -> Right $+ let endtime = case lefts l of+ [] -> Nothing+ ts -> Just (minimum ts)+ in SafeDropProof need mincopies stillhave endtime removallock+ UnsafeDrop -> Left stillhave
Types/Remote.hs view
@@ -2,7 +2,7 @@ - - Most things should not need this, using Types instead -- - Copyright 2011-2021 Joey Hess <id@joeyh.name>+ - Copyright 2011-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -25,6 +25,7 @@ , ExportActions(..) , ImportActions(..) , ByteSize+ , SafeDropProof ) where @@ -105,8 +106,14 @@ , retrievalSecurityPolicy :: RetrievalSecurityPolicy -- Removes a key's contents (succeeds even the contents are not present) -- Can throw exception if unable to access remote, or if remote- -- refuses to remove the content.- , removeKey :: Key -> a ()+ -- refuses to remove the content, or if the proof is expired.+ --+ -- The proof is verified not to have expired shortly+ -- before calling this. But, if the remote's lockContent returns+ -- LockedCopy, the proof's expiry should be checked on the remote,+ -- so that a delay in communicating with the remote does not+ -- cause the removal to happen after the proof expires.+ , removeKey :: Maybe SafeDropProof -> Key -> a () -- Uses locking to prevent removal of a key's contents, -- thus producing a VerifiedCopy, which is passed to the callback. -- If unable to lock, does not run the callback, and throws an
Types/StoreRetrieve.hs view
@@ -10,6 +10,7 @@ module Types.StoreRetrieve where import Annex.Common+import Types.NumCopies import Utility.Metered import Utility.Hash (IncrementalVerifier) @@ -44,8 +45,9 @@ -- Action that removes a Key's content from a remote. -- Succeeds if key is already not present.--- Throws an exception if the remote is not accessible.-type Remover = Key -> Annex ()+-- Throws an exception if the remote is not accessible+-- or the proof has expired.+type Remover = Maybe SafeDropProof -> Key -> Annex () -- Checks if a Key's content is present on a remote. -- Throws an exception if the remote is not accessible.
Types/UUID.hs view
@@ -16,6 +16,7 @@ import Data.Maybe import Data.String import Data.ByteString.Builder+import Control.DeepSeq import qualified Data.Semigroup as Sem import Git.Types (ConfigValue(..))@@ -27,6 +28,10 @@ -- A UUID is either an arbitrary opaque string, or UUID info may be missing. data UUID = NoUUID | UUID B.ByteString deriving (Eq, Ord, Show, Read)++instance NFData UUID where+ rnf NoUUID = ()+ rnf (UUID b) = rnf b class FromUUID a where fromUUID :: UUID -> a
Upgrade/V1.hs view
@@ -1,6 +1,6 @@ {- git-annex v1 -> v2 upgrade support -- - Copyright 2011 Joey Hess <id@joeyh.name>+ - Copyright 2011-2024 Joey Hess <id@joeyh.name> - - Licensed under the GNU AGPL version 3 or higher. -}@@ -18,6 +18,7 @@ import qualified Data.ByteString.Lazy as L import qualified System.FilePath.ByteString as P import System.PosixCompat.Files (isRegularFile)+import Text.Read import Annex.Common import Types.Upgrade@@ -149,13 +150,16 @@ -- WORM backend keys: "WORM:mtime:size:filename" -- all the rest: "backend:key" ----- If the file looks like "WORM:XXX-...", then it was created by mixing+-- If the file looks like "WORM:FOO-...", then it was created by mixing -- v2 and v1; that infelicity is worked around by treating the value -- as the v2 key that it is. readKey1 :: String -> Key-readKey1 v- | mixup = fromJust $ deserializeKey $ intercalate ":" $ Prelude.tail bits- | otherwise = mkKey $ \d -> d+readKey1 = fromMaybe (giveup "unable to parse v0 key") . readKey1'++readKey1' :: String -> Maybe Key+readKey1' v+ | mixup = deserializeKey $ intercalate ":" $ drop 1 bits+ | otherwise = Just $ mkKey $ \d -> d { keyName = S.toShort (encodeBS n) , keyVariety = parseKeyVariety (encodeBS b) , keySize = s@@ -166,13 +170,13 @@ b = Prelude.head bits n = intercalate ":" $ drop (if wormy then 3 else 1) bits t = if wormy- then Just (Prelude.read (bits !! 1) :: EpochTime)+ then readMaybe (bits !! 1) :: Maybe EpochTime else Nothing- s = if wormy- then Just (Prelude.read (bits !! 2) :: Integer)+ s = if wormy && length bits > 2+ then readMaybe (bits !! 2) :: Maybe Integer else Nothing- wormy = Prelude.head bits == "WORM"- mixup = wormy && isUpper (Prelude.head $ bits !! 1)+ wormy = length bits > 1 && headMaybe bits == Just "WORM"+ mixup = wormy && fromMaybe False (isUpper <$> (headMaybe $ bits !! 1)) showKey1 :: Key -> String showKey1 k = intercalate ":" $ filter (not . null)
+ Utility/MonotonicClock.hs view
@@ -0,0 +1,37 @@+{- Monotonic clocks+ -+ - Copyright 2024 Joey Hess <id@joeyh.name>+ -+ - License: BSD-2-clause+ -}++{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE CPP #-}++module Utility.MonotonicClock where++#if MIN_VERSION_clock(0,3,0)+import qualified System.Clock as Clock+#else+import qualified System.Posix.Clock as Clock+#endif+#ifdef linux_HOST_OS+import Utility.Exception+#endif++newtype MonotonicTimestamp = MonotonicTimestamp Integer+ deriving (Show, Eq, Ord, Num)++-- On linux, this uses a clock that advances while the system is suspended,+-- except for on very old kernels (eg 2.6.32).+-- On other systems, that is not available, and the monotonic clock will+-- not advance while suspended.+currentMonotonicTimestamp :: IO MonotonicTimestamp+currentMonotonicTimestamp =+ (MonotonicTimestamp . fromIntegral . Clock.sec) <$>+#ifdef linux_HOST_OS+ (tryNonAsync (Clock.getTime Clock.Boottime)+ >>= either (const $ Clock.getTime Clock.Monotonic) return)+#else+ Clock.getTime Clock.Monotonic+#endif
git-annex.cabal view
@@ -1,5 +1,5 @@ Name: git-annex-Version: 10.20240701+Version: 10.20240731 Cabal-Version: 1.12 License: AGPL-3 Maintainer: Joey Hess <id@joeyh.name>@@ -35,7 +35,6 @@ -- Include only files that are needed make cabal install git-annex work. Extra-Source-Files: stack.yaml- stack-lts-18.13.yaml README CHANGELOG NEWS@@ -173,6 +172,9 @@ Flag Crypton Description: Use the crypton library rather than the no longer maintained cryptonite +Flag Servant+ Description: Use the servant library, enabling using annex+http urls and git-annex p2phttp+ Flag Benchmark Description: Enable benchmarking Default: True@@ -278,7 +280,8 @@ DAV (>= 1.0), network (>= 3.0.0.0), network-bsd,- git-lfs (>= 1.2.0)+ git-lfs (>= 1.2.0),+ clock (>= 0.2.0.0) CC-Options: -Wall GHC-Options: -Wall -fno-warn-tabs -Wincomplete-uni-patterns Default-Language: Haskell2010@@ -311,6 +314,22 @@ else Build-Depends: cryptonite (>= 0.23) + if flag(Servant)+ Build-Depends:+ servant,+ servant-server,+ servant-client,+ servant-client-core,+ warp (>= 3.2.8),+ warp-tls (>= 3.2.2),+ stm (>= 2.5.1)+ CPP-Options: -DWITH_SERVANT+ Other-Modules:+ Command.P2PHttp+ P2P.Http+ P2P.Http.Server+ P2P.Http.State+ if (os(windows)) Build-Depends: Win32 ((>= 2.6.1.0 && < 2.12.0.0) || >= 2.13.4.0),@@ -554,6 +573,7 @@ Annex.Queue Annex.ReplaceFile Annex.RemoteTrackingBranch+ Annex.SafeDropProof Annex.SpecialRemote Annex.SpecialRemote.Config Annex.Ssh@@ -876,6 +896,9 @@ P2P.Address P2P.Annex P2P.Auth+ P2P.Http.Types+ P2P.Http.Client+ P2P.Http.Url P2P.IO P2P.Protocol P2P.Proxy@@ -1008,6 +1031,7 @@ Utility.Base64 Utility.Batch Utility.Bloom+ Utility.MonotonicClock Utility.CoProcess Utility.CopyFile Utility.Daemon
− stack-lts-18.13.yaml
@@ -1,28 +0,0 @@-flags:- git-annex:- production: true- parallelbuild: true- assistant: true- pairing: true- torrentparser: true- magicmime: false- dbus: false- debuglocks: false- benchmark: true- crypton: false-packages:-- '.'-resolver: lts-18.13-extra-deps:-- IfElse-0.85-- aws-0.22-- bloomfilter-2.0.1.0-- git-lfs-1.2.0-- http-client-restricted-0.0.4-- network-multicast-0.3.2-- sandi-0.5-- torrent-10000.1.1-- base16-bytestring-0.1.1.7-- base64-bytestring-1.0.0.3-- bencode-0.6.1.1-- http-client-0.7.9
stack.yaml view
@@ -10,8 +10,9 @@ debuglocks: false benchmark: true crypton: true+ servant: true packages: - '.'-resolver: lts-22.9+resolver: nightly-2024-07-29 extra-deps:-allow-newer: true+- filepath-bytestring-1.4.100.3.2