voidlizard
/
hbs2
mirror of https://github.com/voidlizard/hbs2
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

wip

This commit is contained in:
voidlizard 2025-02-11 19:53:45 +03:00
parent 96b5b051b3
commit deeea55760
5 changed files with 177 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
hbs2-git3/app/GitRemoteHelper.hs
View File

@ -137,31 +137,32 @@ main = flip runContT pure do
setupLogger setupLogger
origStderr <- liftIO $ dup stdError -- origStderr <- liftIO $ dup stdError
(readEnd, writeEnd) <- liftIO createPipe -- (readEnd, writeEnd) <- liftIO createPipe
liftIO $ dupTo writeEnd stdError -- liftIO $ dupTo writeEnd stdError
liftIO $ closeFd writeEnd -- liftIO $ closeFd writeEnd
rStderr <- liftIO $ fdToHandle readEnd -- rStderr <- liftIO $ fdToHandle readEnd
origHandle <- liftIO $ fdToHandle origStderr -- origHandle <- liftIO $ fdToHandle origStderr
liftIO $ hSetBuffering origHandle NoBuffering -- liftIO $ hSetBuffering origHandle NoBuffering
-- liftIO $ IO.hPutStr origHandle "\n" -- liftIO $ IO.hPutStr origHandle "\n"
ContT $ withAsync $ liftIO $ forever do -- ContT $ withAsync $ liftIO $ forever do
-- pause @'Seconds 0.25 -- pause @'Seconds 0.25
wut <- IO.hGetContents rStderr <&> lines -- wut <- IO.hGetContents rStderr <&> lines
for_ wut $ \s -> do -- for_ wut $ \s -> do
IO.hPutStr origHandle (replicate 100 ' ') -- IO.hPutStrLn rStderr s
IO.hPutStr origHandle "\r" -- IO.hPutStr origHandle (replicate 100 ' ')
IO.hPutStr origHandle s -- IO.hPutStr origHandle "\r"
IO.hPutStr origHandle "\r" -- IO.hPutStr origHandle s
pause @'Seconds 0.05 -- IO.hPutStr origHandle "\r"
-- pause @'Seconds 0.05
ContT $ bracket none $ const do -- ContT $ bracket none $ const do
IO.hPutStr origHandle (replicate 100 ' ') -- IO.hPutStr origHandle (replicate 100 ' ')
IO.hPutStr origHandle "\r" -- IO.hPutStr origHandle "\r"
silence -- silence
lift $ void $ installHandler sigPIPE Ignore Nothing lift $ void $ installHandler sigPIPE Ignore Nothing
env <- nullGit3Env env <- nullGit3Env
@ -177,6 +178,7 @@ main = flip runContT pure do
-- d_ <- asks gitRuntimeDict -- d_ <- asks gitRuntimeDict
-- atomically $ writeTVar d_ (Just (RuntimeDict fuck)) -- atomically $ writeTVar d_ (Just (RuntimeDict fuck))
--
conf <- readLocalConf conf <- readLocalConf
@ -191,7 +193,13 @@ main = flip runContT pure do
_ -> pure Nothing _ -> pure Nothing
recover $ connectedDo $ withStateDo do recover $ connectedDo $ withStateDo do
waitRepo Nothing =<< getGitRepoKeyThrow
notice "WAIT-FOR-REPO-DONE"
void $ run dict conf void $ run dict conf
for_ url updateRepoKey for_ url updateRepoKey
flip fix Plain $ \next -> \case flip fix Plain $ \next -> \case

199
hbs2-git3/lib/HBS2/Git3/Import.hs
View File

@ -18,6 +18,7 @@ import HBS2.System.Dir
import Data.Config.Suckless.Almost.RPC import Data.Config.Suckless.Almost.RPC
import Data.Config.Suckless.Script import Data.Config.Suckless.Script
import Control.Applicative
import Codec.Compression.Zlib qualified as Zlib import Codec.Compression.Zlib qualified as Zlib
import Data.ByteString.Lazy.Char8 qualified as LBS8 import Data.ByteString.Lazy.Char8 qualified as LBS8
import Data.ByteString.Lazy qualified as LBS import Data.ByteString.Lazy qualified as LBS
@ -116,8 +117,8 @@ writeAsGitPack dir href = do
data ImportStage = data ImportStage =
ImportStart ImportStart
| ImportWIP Int (Maybe HashRef) | ImportWIP (Timeout 'Seconds) Int (Maybe HashRef)
| ImportWait (Maybe Int) ImportStage | ImportWait (Timeout 'Seconds) (Maybe Int) ImportStage
| ImportDone (Maybe HashRef) | ImportDone (Maybe HashRef)
{- HLINT ignore "Functor law" -} {- HLINT ignore "Functor law" -}
@ -126,130 +127,158 @@ importGitRefLog :: forall m . ( HBS2GitPerks m
) )
=> Git3 m (Maybe HashRef) => Git3 m (Maybe HashRef)
importGitRefLog = withStateDo $ ask >>= \case importGitRefLog = do
Git3Disconnected{} -> throwIO Git3PeerNotConnected
env@Git3Connected{..} -> do
packs <- gitDir packs <- gitDir
>>= orThrow NoGitDir >>= orThrow NoGitDir
<&> (</> "objects/pack") <&> (</> "objects/pack")
mkdir packs mkdir packs
sto <- getStorage doImport packs `catch` (\( e :: OperationError) -> err (viaShow e) >> pause @'Seconds 1 >> doImport packs)
already_ <- newTVarIO (mempty :: HashSet HashRef) where
doImport packs = withStateDo $ ask >>= \case
Git3Disconnected{} -> throwIO Git3PeerNotConnected
Git3Connected{..} -> flip runContT pure do
flip fix ImportStart $ \again -> \case sto <- getStorage
ImportDone x -> do
notice "import done"
updateReflogIndex
for_ x updateImportedCheckpoint
refs <- importedRefs already_ <- newTVarIO (mempty :: HashSet HashRef)
if not (null refs && isJust x) then do oldRvl <- gitRefLogVal & readTVarIO
pure x reflog <- getGitRemoteKey >>= orThrow Git3ReflogNotSet
else do newRvl_ <- newTVarIO Nothing
notice $ "no refs arrived - go again"
again ImportStart
ImportWait d next -> do void $ ContT $ withAsync $ forever do
void $ lift (callRpcWaitMay @RpcRefLogFetch (TimeoutSec 2) reflogAPI reflog)
pause @'Seconds 1.15 lift (callRpcWaitMay @RpcRefLogGet (TimeoutSec 2) reflogAPI reflog)
>>= \case
Just (Just x) | Just x /= oldRvl -> atomically (writeTVar newRvl_ (Just x))
_ -> none
down <- callRpcWaitRetry @RpcGetProbes (TimeoutSec 1) 3 peerAPI () pause @'Seconds 10
>>= orThrow RpcTimeout
<&> maybe 0 fromIntegral . headMay . mapMaybe \case
ProbeSnapshotElement "Download.wip" n -> Just n
_ -> Nothing
notice $ "wait some time..." <+> parens (pretty down) lift $ flip fix ImportStart $ \again -> \case
ImportDone x -> do
notice "import done"
case d of newRlv <- readTVarIO newRvl_
Just n | down /= n || down == 0 -> again next let doAgain = newRlv /= oldRvl
_ -> pause @'Seconds 2.85 >> again (ImportWait (Just down) next) updateReflogIndex
for_ x updateImportedCheckpoint
ImportStart -> do refs <- importedRefs
rvl <- readTVarIO gitRefLogVal if not (null refs && isJust x) || doAgain then do
pure x
else do
atomically do
writeTVar newRvl_ Nothing
writeTVar gitRefLogVal (newRlv <|> oldRvl)
importGroupKeys notice $ "import: go again"
again ImportStart
prev <- importedCheckpoint ImportWait sec d next -> do
if | isNothing prev -> again $ ImportWIP 0 prev pause sec
| prev /= rvl -> do down <- callRpcWaitRetry @RpcGetProbes (TimeoutSec 1) 3 peerAPI ()
again $ ImportWIP 0 prev >>= orThrow RpcTimeout
<&> maybe 0 fromIntegral . headMay . mapMaybe \case
ProbeSnapshotElement "Download.wip" n -> Just n
_ -> Nothing
| otherwise -> again $ ImportDone prev notice $ "wait some time..." <+> parens (pretty down)
ImportWIP attempt prev -> do case d of
Just n | down /= n || down == 0 -> again next
notice $ "download wip" <+> pretty attempt _ -> pause @'Seconds 2.85 >> again (ImportWait (sec*1.10) (Just down) next)
r <- try @_ @OperationError $ do ImportStart -> do
excl <- maybe1 prev (pure mempty) $ \p -> do rvl <- readTVarIO gitRefLogVal
txListAll (Just p) <&> HS.fromList . fmap fst
rv <- refLogRef importGroupKeys
hxs <- txList ( pure . not . flip HS.member excl ) rv prev <- importedCheckpoint
cp' <- flip fix (fmap snd hxs, Nothing) $ \next -> \case if | isNothing prev -> again $ ImportWIP 1.0 0 prev
([], r) -> pure (gitTxTree <$> r)
(TxSegment{}:xs, l) -> next (xs, l)
(cp@(TxCheckpoint n tree) : xs, l) -> do
-- full <- findMissedBlocks sto tree <&> L.null | prev /= rvl -> do
missed_ <- newTVarIO 0 again $ ImportWIP 1.0 0 prev
deepScan ScanDeep (\_ -> atomically $ modifyTVar missed_ succ)
(coerce tree)
(getBlock sto)
(const none)
full <- readTVarIO missed_ <&> (==0) | otherwise -> again $ ImportDone prev
if full && Just n > (getGitTxRank <$> l) then do ImportWIP w attempt prev -> do
next (xs, Just cp)
else do
next (xs, l)
case cp' of notice $ "download wip" <+> pretty attempt
Nothing -> do
notice "no checkpoints found"
pure Nothing
Just cp -> do r <- try @_ @OperationError $ do
notice $ "found checkpoint" <+> pretty cp excl <- maybe1 prev (pure mempty) $ \p -> do
txs <- txList ( pure . not . flip HS.member excl ) (Just cp) txListAll (Just p) <&> HS.fromList . fmap fst
forConcurrently_ txs $ \case rv <- refLogRef
(_, TxCheckpoint{}) -> none
(h, TxSegment tree) -> do
new <- readTVarIO already_ <&> not . HS.member tree
when new do hxs <- txList ( pure . not . flip HS.member excl ) rv
s <- writeAsGitPack packs tree
for_ s $ \file -> do cp' <- flip fix (fmap snd hxs, Nothing) $ \next -> \case
gitRunCommand [qc|git index-pack {file}|] ([], r) -> pure (gitTxTree <$> r)
>>= orThrowPassIO (TxSegment{}:xs, l) -> next (xs, l)
(cp@(TxCheckpoint n tree) : xs, l) -> do
atomically $ modifyTVar already_ (HS.insert tree) missed <- findMissedBlocks sto tree
notice $ "imported" <+> pretty h
pure (Just cp) let full = L.null missed
case r of if full && Just n > (getGitTxRank <$> l) then do
Right cp -> again $ ImportDone cp next (xs, Just cp)
Left (MissedBlockError2 _) -> notice "missed blocks" >> again (ImportWait Nothing (ImportWIP (succ attempt) prev)) else do
Left MissedBlockError -> notice "missed blocks" >> again (ImportWait Nothing (ImportWIP (succ attempt) prev)) next (xs, l)
Left e -> err (viaShow e) >> throwIO e
case cp' of
Nothing -> do
notice "no checkpoints found"
pure Nothing
Just cp -> do
notice $ "found checkpoint" <+> pretty cp
txs <- txList ( pure . not . flip HS.member excl ) (Just cp)
forConcurrently_ txs $ \case
(_, TxCheckpoint{}) -> none
(h, TxSegment tree) -> do
new <- readTVarIO already_ <&> not . HS.member tree
when new do
s <- writeAsGitPack packs tree
for_ s $ \file -> do
gitRunCommand [qc|git index-pack {file}|]
>>= orThrowPassIO
atomically $ modifyTVar already_ (HS.insert tree)
notice $ "imported" <+> pretty h
pure (Just cp)
case r of
Right cp -> again $ ImportDone cp
Left (MissedBlockError2 _) -> do
notice "missed blocks"
again (ImportWait w Nothing (ImportWIP (w*1.15) (succ attempt) prev))
Left MissedBlockError -> do
notice "missed blocks"
again (ImportWait w Nothing (ImportWIP (w*1.15) (succ attempt) prev))
Left e -> err (viaShow e) >> throwIO e
groupKeysFile :: (MonadIO m) => Git3 m FilePath groupKeysFile :: (MonadIO m) => Git3 m FilePath
@ -271,7 +300,7 @@ importGroupKeys :: forall m . ( HBS2GitPerks m
importGroupKeys = do importGroupKeys = do
debug $ "importGroupKeys" notice $ "importGroupKeys"
sto <- getStorage sto <- getStorage
already <- readGroupKeyFile already <- readGroupKeyFile

2
hbs2-git3/lib/HBS2/Git3/Run.hs
View File

@ -566,7 +566,7 @@ compression ; prints compression level
entry $ bindMatch "repo:relay-only" $ nil_ $ \case entry $ bindMatch "repo:relay-only" $ nil_ $ \case
[ SignPubKeyLike repo ] -> lift $ connectedDo do [ SignPubKeyLike repo ] -> lift $ connectedDo do
setGitRepoKey repo setGitRepoKey repo
waitRepo (Just 2) =<< getGitRepoKeyThrow waitRepo (Just 3) =<< getGitRepoKeyThrow
_ -> throwIO (BadFormException @C nil) _ -> throwIO (BadFormException @C nil)

74
hbs2-git3/lib/HBS2/Git3/State.hs
View File

@ -256,8 +256,6 @@ waitRepo timeout repoKey = do
when (rlv && rlog) $ done () when (rlv && rlog) $ done ()
reflog_ <- newEmptyTMVarIO
let wait w what x = pause @'Seconds w >> what x let wait w what x = pause @'Seconds w >> what x
callCC \forPeer -> do callCC \forPeer -> do
@ -271,67 +269,63 @@ waitRepo timeout repoKey = do
void (callRpcWaitMay @RpcLWWRefFetch (TimeoutSec 1) lwwAPI (LWWRefKey repoKey)) void (callRpcWaitMay @RpcLWWRefFetch (TimeoutSec 1) lwwAPI (LWWRefKey repoKey))
pause @'Seconds 10 pause @'Seconds 10
pFetchRefLog <- ContT $ withAsync do lww <- flip fix 2 \next i -> do
r <- atomically $ takeTMVar reflog_
forever do
void (callRpcWaitMay @RpcRefLogFetch (TimeoutSec 1) reflogAPI r)
pause @'Seconds 10
lww <- flip fix () \next _ -> do
notice $ "wait for" <+> pretty (AsBase58 repoKey) notice $ "wait for" <+> pretty (AsBase58 repoKey)
lift (callRpcWaitMay @RpcLWWRefGet (TimeoutSec 1) lwwAPI (LWWRefKey repoKey)) lift (callRpcWaitMay @RpcLWWRefGet (TimeoutSec 1) lwwAPI (LWWRefKey repoKey))
>>= \case >>= \case
Just (Just x) -> pure x Just (Just x) -> pure x
_ -> wait 2 next () _ -> wait i next (i*1.05)
setGitRepoKey repoKey setGitRepoKey repoKey
notice $ "lwwref value" <+> pretty (lwwValue lww) notice $ "lwwref value" <+> pretty (lwwValue lww)
mf <- flip fix () $ \next _ -> do mf <- flip fix 3 $ \next i -> do
notice $ "wait for manifest" notice $ "wait for manifest" <+> pretty i
lift (try @_ @WalkMerkleError getRepoManifest) >>= \case lift (try @_ @SomeException getRepoManifest) >>= \case
Left{} -> wait 1 next () Left{} -> wait i next (i*1.10)
Right x -> pure x Right x -> pure x
reflog <- getRefLog mf & orThrow GitRepoManifestMalformed reflog <- getRefLog mf & orThrow GitRepoManifestMalformed
atomically $ writeTMVar reflog_ reflog
lift (callRpcWaitMay @RpcPollAdd (TimeoutSec 1) peerAPI (reflog, "reflog", 11)) lift (callRpcWaitMay @RpcPollAdd (TimeoutSec 1) peerAPI (reflog, "reflog", 11))
>>= orThrow RpcTimeout >>= orThrow RpcTimeout
rv <- flip fix () \next _ -> do let waiter = maybe (forever (pause @'Seconds 3600)) pause timeout
notice $ "wait for data" <+> pretty (AsBase58 reflog)
lift (callRpcWaitMay @RpcRefLogGet (TimeoutSec 1) reflogAPI reflog)
>>= \case
Just (Just x) -> pure x
_ -> wait 2 next ()
atomically $ writeTVar gitRefLogVal (Just rv) ContT $ withAsync $ do
pause @'Seconds 1
flip fix 2 $ \next i -> do
debug $ "fetch reflog" <+> pretty (AsBase58 reflog)
void $ lift (callRpcWaitMay @RpcRefLogFetch (TimeoutSec 2) reflogAPI reflog)
pause @'Seconds i
next (i*1.05)
okay <- newEmptyTMVarIO
flip fix () $ \next _ -> do void $ lift $ race waiter do
notice $ "wait for data (2)" <+> pretty (AsBase58 reflog)
-- missed <- findMissedBlocks sto rv
missed_ <- newTVarIO 0
lift $ deepScan ScanDeep (\_ -> atomically $ modifyTVar missed_ succ) (coerce rv) (getBlock sto) (const none)
missed <- readTVarIO missed_
when (missed > 0) do rv <- flip fix 1 \next i -> do
notice $ "still missed blocks:" <+> pretty missed notice $ "wait for reflog" <+> pretty i <+> pretty (AsBase58 reflog)
wait 5 next () lift (callRpcWaitMay @RpcRefLogGet (TimeoutSec 2) reflogAPI reflog)
>>= \case
Just (Just x) -> pure x
Nothing -> debug "fucking RPC timeout!" >> wait i next (i*1.05)
_ -> wait i next (i*1.05)
atomically $ writeTMVar okay True atomically $ writeTVar gitRefLogVal (Just rv)
pWait <- ContT $ withAsync $ race ( pause (fromMaybe 300 timeout) ) do cancel pFetch
void $ atomically $ takeTMVar okay
waitAnyCatchCancel [pWait, pFetch, pFetchRefLog] notice $ "reflog" <+> pretty (AsBase58 reflog) <+> pretty rv
lift $ updateRepoKey repoKey flip fix 5 $ \next w -> do
debug $ "reflog" <+> pretty (AsBase58 reflog) <+> pretty rv handle (\(e :: OperationError) -> pause @'Seconds w >> next (w*1.10)) do
missed <- findMissedBlocks sto rv
if L.null missed then do
updateRepoKey repoKey
else do
notice $ "wait reflog to sync in consistent state" <+> pretty w
pause @'Seconds w
next (w*1.01)

2
hbs2-peer/app/PeerMain.hs
View File

@ -832,7 +832,7 @@ runPeer opts = respawnOnError opts $ do
stn <- getNumCapabilities <&> max 2 . div 4 stn <- getNumCapabilities <&> max 2 . div 4
w <- replicateM 2 $ async $ liftIO $ simpleStorageWorker s w <- replicateM stn $ async $ liftIO $ simpleStorageWorker s
localMulticast <- liftIO $ (headMay <$> parseAddrUDP (fromString defLocalMulticast) localMulticast <- liftIO $ (headMay <$> parseAddrUDP (fromString defLocalMulticast)
<&> fmap (fromSockAddr @'UDP . addrAddress) ) <&> fmap (fromSockAddr @'UDP . addrAddress) )