mirror of https://github.com/voidlizard/hbs2
hbs2-peer migrate from NCQv1, test
This commit is contained in:
voidlizard
parent
513f03eeb3
commit
789c798e7e
|
|
@ -4,69 +4,20 @@ module HBS2.CLI.NCQ3.Migrate where
|
|||
import HBS2.CLI.Prelude
|
||||
import HBS2.CLI.Run.Internal
|
||||
|
||||
import HBS2.Base58
|
||||
import HBS2.Data.Types.Refs
|
||||
import HBS2.Hash
|
||||
import HBS2.Peer.NCQ3.Migrate.NCQ
|
||||
|
||||
import HBS2.Net.Auth.Schema()
|
||||
import HBS2.Peer.CLI.Detect
|
||||
import HBS2.Peer.RPC.API.LWWRef
|
||||
import HBS2.Peer.RPC.API.Peer
|
||||
import HBS2.Peer.RPC.API.RefLog
|
||||
import HBS2.Peer.RPC.API.Storage
|
||||
import HBS2.Peer.RPC.Client
|
||||
import HBS2.Peer.RPC.Client.Unix
|
||||
import HBS2.Storage
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.NCQ3.Internal.Files
|
||||
import HBS2.Storage.NCQ3.Internal.Fossil
|
||||
import HBS2.Storage.NCQ3.Internal.Index
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||
import HBS2.Peer.Proto.RefLog
|
||||
import HBS2.Peer.Proto.RefChan
|
||||
import HBS2.Peer.Proto.LWWRef
|
||||
|
||||
import Data.Config.Suckless.System
|
||||
|
||||
import Data.ByteString.Lazy qualified as LBS
|
||||
import Data.ByteString qualified as BS
|
||||
import Control.Monad.Trans.Cont
|
||||
import Control.Monad.Trans.Maybe
|
||||
import Data.HashSet (HashSet)
|
||||
import Data.HashSet qualified as HS
|
||||
import Data.HashMap.Strict qualified as HM
|
||||
import Data.HashMap.Strict (HashMap)
|
||||
import Data.Either
|
||||
import Data.List qualified as L
|
||||
import System.Directory (getModificationTime)
|
||||
import UnliftIO.IO
|
||||
|
||||
import Streaming.Prelude qualified as S
|
||||
|
||||
{- HLINT ignore "Functor law"-}
|
||||
|
||||
data SomeRef = forall a . ( Eq a
|
||||
, Hashable a
|
||||
, Hashed HbSync a
|
||||
, Pretty a
|
||||
, RefMetaData a
|
||||
)
|
||||
=> SomeRef a
|
||||
|
||||
instance RefMetaData SomeRef where
|
||||
refMetaData (SomeRef a) = refMetaData a
|
||||
|
||||
instance Pretty SomeRef where
|
||||
pretty (SomeRef wtf) = pretty wtf
|
||||
|
||||
instance Hashed HbSync SomeRef where
|
||||
hashObject (SomeRef a) = hashObject a
|
||||
|
||||
legacyHashRefNCQ1 :: HashRef -> HashRef -> HashRef
|
||||
legacyHashRefNCQ1 salt h = HashRef (hashObject (coerce @_ @ByteString h <> coerce salt))
|
||||
|
||||
legacyHashRefFor :: Hashed HbSync p => HashRef -> p -> HashRef
|
||||
legacyHashRefFor salt x = legacyHashRefNCQ1 salt (HashRef $ hashObject @HbSync $ x)
|
||||
|
||||
migrateEntries :: forall c m . ( MonadUnliftIO m
|
||||
, IsContext c
|
||||
, Exception (BadFormException c)
|
||||
|
|
@ -92,9 +43,9 @@ migrateEntries = do
|
|||
<&> fromMaybe mempty
|
||||
|
||||
rrefs <- S.toList_ <$> for refs $ \(pk, s, _) -> case s of
|
||||
"reflog" -> S.yield (SomeRef $ RefLogKey @'HBS2Basic pk)
|
||||
"refchan" -> S.yield (SomeRef $ RefChanLogKey @'HBS2Basic pk)
|
||||
"lwwref" -> S.yield (SomeRef $ LWWRefKey @'HBS2Basic pk)
|
||||
"reflog" -> S.yield (WrapRef $ RefLogKey @'HBS2Basic pk)
|
||||
"refchan" -> S.yield (WrapRef $ RefChanLogKey @'HBS2Basic pk)
|
||||
"lwwref" -> S.yield (WrapRef $ LWWRefKey @'HBS2Basic pk)
|
||||
_ -> none
|
||||
|
||||
lift $ migrateNCQ1 nicelog rrefs src dst
|
||||
|
|
@ -104,90 +55,3 @@ migrateEntries = do
|
|||
nicelog :: forall m . MonadIO m => Doc AnsiStyle -> m ()
|
||||
nicelog doc = liftIO $ hPutDoc stdout (doc <> line)
|
||||
|
||||
migrateNCQ1 :: MonadUnliftIO m
|
||||
=> ( Doc AnsiStyle -> IO () )
|
||||
-> [SomeRef]
|
||||
-> FilePath
|
||||
-> FilePath
|
||||
-> m ()
|
||||
migrateNCQ1 logger refs src b = flip runContT pure do
|
||||
debug "migrate NCQ to NCQ3"
|
||||
|
||||
let a0 = src </> "0"
|
||||
a0here <- doesPathExist a0
|
||||
|
||||
let a = if a0here then a0 else src
|
||||
|
||||
ncq <- ContT $ ncqWithStorage b
|
||||
|
||||
let sto = AnyStorage ncq
|
||||
|
||||
let mlog = ncqGetFileName ncq "migrate-ncq1.log"
|
||||
|
||||
info $ "set SOURCE dir" <+> pretty a
|
||||
|
||||
ncqSalt <- liftIO (try @_ @IOException (BS.readFile (src </> ".seed")))
|
||||
<&> fromRight mempty
|
||||
<&> HashRef . hashObject
|
||||
|
||||
notice $ "ref. salt" <+> pretty ncqSalt
|
||||
|
||||
files <- do
|
||||
fs <- dirFiles a <&> L.filter (\x -> takeExtension x == ".data")
|
||||
mtimes <- liftIO $ mapM getModificationTime fs
|
||||
pure [ f | (f,_) <- L.sortOn snd (zip fs mtimes) ]
|
||||
|
||||
touch mlog
|
||||
|
||||
processed <- HS.fromList . lines <$> liftIO (readFile mlog)
|
||||
|
||||
let refz = HM.fromList [ (legacyHashRefFor ncqSalt r, SomeRef r) | SomeRef r <- refs ]
|
||||
|
||||
rv <- newTVarIO ( mempty :: HashMap HashRef (SomeRef, HashRef) )
|
||||
nblk <- newTVarIO 0
|
||||
|
||||
for_ files $ \f -> void $ runMaybeT do
|
||||
|
||||
guard ( not $ HS.member f processed )
|
||||
|
||||
debug $ "processing" <+> pretty f
|
||||
|
||||
ncqStorageScanDataFile0 f $ \offset w key s -> void $ runMaybeT do
|
||||
(t, bs) <- toMPlus (ncqEntryUnwrapValue s)
|
||||
|
||||
case t of
|
||||
B -> do
|
||||
h1 <- putBlock sto (LBS.fromStrict bs)
|
||||
atomically $ modifyTVar nblk succ
|
||||
|
||||
R -> do
|
||||
case HM.lookup key refz of
|
||||
Nothing -> none
|
||||
Just r -> do
|
||||
atomically $ modifyTVar rv (HM.insert (HashRef $ hashObject r) (r, HashRef (coerce bs)))
|
||||
|
||||
_ -> none
|
||||
|
||||
|
||||
liftIO $ appendFile mlog (f <> "\n")
|
||||
|
||||
nb <- readTVarIO nblk
|
||||
liftIO $ logger $ "moved" <+> pretty nb <+> "blocks"
|
||||
|
||||
foundRefs <- readTVarIO rv <&> HM.toList
|
||||
for_ foundRefs $ \(_, (r, v)) -> do
|
||||
blk <- hasBlock sto (coerce v)
|
||||
when (isJust blk) do
|
||||
liftIO $ logger $ green "update ref" <+> pretty r <+> pretty v
|
||||
updateRef sto r (coerce v)
|
||||
|
||||
-- for_ refs $ \(SomeRef r) -> void $ runMaybeT do
|
||||
-- v <- getRef origSto r >>= toMPlus
|
||||
-- v0 <- hasBlock origSto v
|
||||
-- v1 <- hasBlock sto v
|
||||
-- updateRef sto r v
|
||||
-- v3 <- getRef sto r
|
||||
-- let okay = if v3 == Just v then green "ok" else red "fail"
|
||||
-- liftIO $ logger $ okay <+> "update ref" <+> pretty r <+> pretty v <+> pretty v1 <+> parens (pretty v0)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,14 +1,25 @@
|
|||
module Migrate where
|
||||
|
||||
import HBS2.Prelude.Plated
|
||||
import HBS2.Clock
|
||||
import HBS2.Misc.PrettyStuff
|
||||
import HBS2.Hash
|
||||
import HBS2.Data.Types.Refs
|
||||
import HBS2.OrDie
|
||||
import HBS2.Defaults
|
||||
import HBS2.Storage.NCQ
|
||||
import HBS2.Storage
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Peer.Proto.RefLog
|
||||
import HBS2.Peer.Proto.RefChan
|
||||
import HBS2.Peer.Proto.LWWRef
|
||||
|
||||
import HBS2.Net.Proto.Types
|
||||
import Log
|
||||
import PeerConfig
|
||||
import Brains
|
||||
|
||||
import HBS2.Peer.NCQ3.Migrate.NCQ qualified as N
|
||||
import HBS2.Peer.NCQ3.Migrate.NCQ (WrapRef(..))
|
||||
|
||||
import Data.Config.Suckless.Script hiding (optional)
|
||||
import Data.Config.Suckless.Script.File (glob)
|
||||
|
|
@ -26,262 +37,330 @@ import System.Exit
|
|||
import Data.ByteString qualified as BS
|
||||
import Data.ByteString.Lazy qualified as LBS
|
||||
import Control.Exception
|
||||
import Control.Monad.Cont
|
||||
import Control.Monad.Trans.Cont
|
||||
import Control.Monad.Reader
|
||||
|
||||
import UnliftIO
|
||||
-- import UnliftIO.Temporary
|
||||
|
||||
import Streaming.Prelude qualified as S
|
||||
|
||||
migrate :: [Syntax C]-> IO ()
|
||||
migrate syn = flip runContT pure $ callCC \exit -> do
|
||||
|
||||
xdg <- liftIO $ getXdgDirectory XdgData defStorePath <&> fromString
|
||||
|
||||
|
||||
let (opts, argz) = splitOpts [ ("-n",0)
|
||||
, ("--dry",0)
|
||||
, ("--no-refs",0)
|
||||
, ("--help",0)
|
||||
let (opts, argz) = splitOpts [ ("-c",1)
|
||||
-- , ("--dry",0)
|
||||
-- , ("--no-refs",0)
|
||||
-- , ("--help",0)
|
||||
] syn
|
||||
|
||||
prefix <- headMay [ p | StringLike p <- argz ]
|
||||
& orThrowUser ( "Storage dir not specified" <+> parens ("typically" <+> pretty xdg) <> line
|
||||
<> line
|
||||
<> "run hbs2-peer migrate" <+> pretty xdg <> line
|
||||
<> "if this is it"
|
||||
)
|
||||
-- FIXME: migrate-simple-storage!
|
||||
-- KISS. just import block/remove block.
|
||||
-- let the user backup it.
|
||||
--
|
||||
--
|
||||
--
|
||||
|
||||
conf@(PeerConfig se) <- peerConfigRead (headMay [p | ListVal [StringLike "-c", StringLike p] <- opts] )
|
||||
|
||||
let dry = or [ True | ListVal [StringLike s] <- opts, s `elem` ["--dry","-n"]]
|
||||
brains <- newBasicBrains conf
|
||||
|
||||
let norefs = or [ True | ListVal [StringLike "--no-refs"] <- opts ]
|
||||
bProbe <- newSimpleProbe "Brains"
|
||||
brainsThread <- ContT $ withAsync $ runBasicBrains conf brains
|
||||
|
||||
xdg <- liftIO $ getXdgDirectory XdgData defStorePath <&> fromString
|
||||
|
||||
let prefix = fromMaybe xdg $ runReader (cfgValue @PeerStorageKey @(Maybe FilePath)) se
|
||||
|
||||
let store = prefix
|
||||
|
||||
let migrateDir = store </> "migrate"
|
||||
let ncqDir = store </> "ncq"
|
||||
let ncqDirBackup = store </> ".ncq.backup"
|
||||
let ncq3Dir = store </> "ncq3"
|
||||
|
||||
liftIO $ IO.hSetBuffering stdin NoBuffering
|
||||
liftIO $ IO.hSetBuffering stdout LineBuffering
|
||||
ncqHere <- Sy.doesDirectoryExist ncqDir
|
||||
|
||||
already <- Sy.doesDirectoryExist migrateDir
|
||||
unless ncqHere $ exit ()
|
||||
|
||||
when already do
|
||||
liftIO $ hPutDoc stdout $ yellow "Found migration WIP" <+> pretty migrateDir <> "," <+> "continue" <> line
|
||||
liftIO $ hPrint stderr $
|
||||
"Migrate" <+> pretty ncqDir <> line
|
||||
<> "you may remove" <+> pretty ncqDir
|
||||
<+> "when migration successfully done"
|
||||
<+> "or you may back it up" <> line
|
||||
|
||||
liftIO $ hPutDoc stdout $
|
||||
yellow "Storage migration process is about to start" <> line
|
||||
<> "It will convert the current storage structure to a new one (NCQ storage)" <> line
|
||||
<> "to use with hbs2 0.25.2 and newer" <> line
|
||||
<> "hbs2-peer 0.25.1 and earlier versions don't work with the new storage." <> line
|
||||
<> "If you want to backup your data first just for in case" <> line
|
||||
<> "You may store the contents of directory" <> line
|
||||
<> line
|
||||
<> pretty prefix
|
||||
<> line <> line
|
||||
<> "specifically" <+> pretty (prefix </> "blocks") <+> "and" <+> pretty (prefix</> "refs")
|
||||
<> line
|
||||
<> "to roll back to the older version --- just restore them"
|
||||
<> line
|
||||
flip fix 10 \next i -> do
|
||||
liftIO $ hPrint stderr $ pretty i <> "..."
|
||||
pause @'Seconds 1
|
||||
when (i > 0) $ next (pred i)
|
||||
|
||||
liftIO do
|
||||
IO.hFlush stdout
|
||||
IO.hFlush stderr
|
||||
putStr "Start the migration process? [y]: "
|
||||
IO.hFlush stdout
|
||||
notice "Go!"
|
||||
|
||||
y <- liftIO getChar
|
||||
notice "Seek for polled references"
|
||||
|
||||
unless ( toUpper y == 'Y' ) $ exit ()
|
||||
refs <- listPolledRefs @L4Proto brains Nothing
|
||||
|
||||
liftIO do
|
||||
putStrLn ""
|
||||
rrefs <- S.toList_ <$> for refs $ \(pk, s, _) -> case s of
|
||||
"reflog" -> S.yield (WrapRef $ RefLogKey @'HBS2Basic pk)
|
||||
"refchan" -> S.yield (WrapRef $ RefChanLogKey @'HBS2Basic pk)
|
||||
"lwwref" -> S.yield (WrapRef $ LWWRefKey @'HBS2Basic pk)
|
||||
_ -> none
|
||||
|
||||
info $ "migration started" <+> pretty opts
|
||||
|
||||
info $ "create dir" <+> pretty migrateDir
|
||||
notice $ "got references" <+> vcat (pretty <$> rrefs)
|
||||
|
||||
mkdir migrateDir
|
||||
lift $ N.migrateNCQ1 notice rrefs ncqDir ncq3Dir
|
||||
|
||||
wip <- Sy.doesDirectoryExist migrateDir
|
||||
notice $ "move" <+> pretty ncqDir <+> pretty ncqDirBackup
|
||||
<> line <> "you may remove it if you want"
|
||||
|
||||
source <- if dry && not wip then do
|
||||
pure store
|
||||
else do
|
||||
info $ yellow "Real migration," <+> "fasten the sit belts!"
|
||||
let srcDir = migrateDir </> "source"
|
||||
mkdir srcDir
|
||||
let inBlk = store </> "blocks"
|
||||
let inRefs = store </> "refs"
|
||||
mv ncqDir ncqDirBackup
|
||||
|
||||
e1 <- Sy.doesDirectoryExist inBlk
|
||||
|
||||
when e1 $ mv inBlk (srcDir </> "blocks")
|
||||
-- let (opts, argz) = splitOpts [ ("-n",0)
|
||||
-- , ("--dry",0)
|
||||
-- , ("--no-refs",0)
|
||||
-- , ("--help",0)
|
||||
-- ] syn
|
||||
|
||||
e2 <- Sy.doesDirectoryExist inRefs
|
||||
-- prefix <- headMay [ p | StringLike p <- argz ]
|
||||
-- & orThrowUser ( "Storage dir not specified" <+> parens ("typically" <+> pretty xdg) <> line
|
||||
-- <> line
|
||||
-- <> "run hbs2-peer migrate" <+> pretty xdg <> line
|
||||
-- <> "if this is it"
|
||||
-- )
|
||||
|
||||
when e2 $ mv inRefs (srcDir </> "refs")
|
||||
|
||||
pure srcDir
|
||||
-- let dry = or [ True | ListVal [StringLike s] <- opts, s `elem` ["--dry","-n"]]
|
||||
|
||||
tmp <- ContT $ Temp.withTempDirectory migrateDir "run"
|
||||
-- let norefs = or [ True | ListVal [StringLike "--no-refs"] <- opts ]
|
||||
|
||||
info $ "create dir" <+> pretty tmp
|
||||
|
||||
let blkz = source </> "blocks"
|
||||
let refz = source </> "refs"
|
||||
-- let store = prefix
|
||||
|
||||
let b = tmp </> "blocks"
|
||||
let r = tmp </> "refs"
|
||||
-- let migrateDir = store </> "migrate"
|
||||
-- let ncqDir = store </> "ncq"
|
||||
|
||||
info $ "create directory links"
|
||||
info $ pretty blkz <+> pretty b
|
||||
liftIO $ createDirectoryLink blkz b
|
||||
-- liftIO $ IO.hSetBuffering stdin NoBuffering
|
||||
-- liftIO $ IO.hSetBuffering stdout LineBuffering
|
||||
|
||||
info $ pretty refz <+> pretty r
|
||||
liftIO $ createDirectoryLink refz r
|
||||
-- already <- Sy.doesDirectoryExist migrateDir
|
||||
|
||||
ncq <- ContT $ withNCQ id ncqDir
|
||||
-- when already do
|
||||
-- liftIO $ hPutDoc stdout $ yellow "Found migration WIP" <+> pretty migrateDir <> "," <+> "continue" <> line
|
||||
|
||||
let nameToHash fn =
|
||||
fromString @HashRef $ mconcat $ reverse $ take 2 $ reverse $ splitDirectories fn
|
||||
-- liftIO $ hPutDoc stdout $
|
||||
-- yellow "Storage migration process is about to start" <> line
|
||||
-- <> "It will convert the current storage structure to a new one (NCQ storage)" <> line
|
||||
-- <> "to use with hbs2 0.25.2 and newer" <> line
|
||||
-- <> "hbs2-peer 0.25.1 and earlier versions don't work with the new storage." <> line
|
||||
-- <> "If you want to backup your data first just for in case" <> line
|
||||
-- <> "You may store the contents of directory" <> line
|
||||
-- <> line
|
||||
-- <> pretty prefix
|
||||
-- <> line <> line
|
||||
-- <> "specifically" <+> pretty (prefix </> "blocks") <+> "and" <+> pretty (prefix</> "refs")
|
||||
-- <> line
|
||||
-- <> "to roll back to the older version --- just restore them"
|
||||
-- <> line
|
||||
|
||||
let hashToPath ha = do
|
||||
let (p,r) = splitAt 1 (show $ pretty ha)
|
||||
p </> r
|
||||
-- liftIO do
|
||||
-- IO.hFlush stdout
|
||||
-- IO.hFlush stderr
|
||||
-- putStr "Start the migration process? [y]: "
|
||||
-- IO.hFlush stdout
|
||||
|
||||
checkQ <- newTQueueIO
|
||||
checkN <- newTVarIO 0
|
||||
-- y <- liftIO getChar
|
||||
|
||||
errors <- newTVarIO 0
|
||||
-- unless ( toUpper y == 'Y' ) $ exit ()
|
||||
|
||||
rmp <- liftIO $ async $ fix \next -> do
|
||||
atomically (readTQueue checkQ) >>= \case
|
||||
Nothing -> none
|
||||
Just what -> do
|
||||
-- liftIO do
|
||||
-- putStrLn ""
|
||||
|
||||
toWipe <- ncqLocate ncq what >>= \case
|
||||
Just (InCurrent{}) -> do
|
||||
atomically $ modifyTVar checkN pred
|
||||
pure True
|
||||
-- info $ "migration started" <+> pretty opts
|
||||
|
||||
Just (InFossil{}) -> do
|
||||
atomically $ modifyTVar checkN pred
|
||||
pure True
|
||||
-- info $ "create dir" <+> pretty migrateDir
|
||||
|
||||
Just (InWriteQueue{}) -> do
|
||||
atomically $ unGetTQueue checkQ (Just what)
|
||||
pure False
|
||||
-- mkdir migrateDir
|
||||
|
||||
Nothing -> do
|
||||
atomically $ modifyTVar errors succ
|
||||
pure False
|
||||
-- wip <- Sy.doesDirectoryExist migrateDir
|
||||
|
||||
when toWipe do
|
||||
let path = b </> hashToPath what
|
||||
info $ yellow "d" <+> pretty what
|
||||
-- source <- if dry && not wip then do
|
||||
-- pure store
|
||||
-- else do
|
||||
-- info $ yellow "Real migration," <+> "fasten the sit belts!"
|
||||
-- let srcDir = migrateDir </> "source"
|
||||
-- mkdir srcDir
|
||||
-- let inBlk = store </> "blocks"
|
||||
-- let inRefs = store </> "refs"
|
||||
|
||||
unless dry do
|
||||
rm path
|
||||
-- e1 <- Sy.doesDirectoryExist inBlk
|
||||
|
||||
next
|
||||
-- when e1 $ mv inBlk (srcDir </> "blocks")
|
||||
|
||||
cnt <- newTVarIO 0
|
||||
-- e2 <- Sy.doesDirectoryExist inRefs
|
||||
|
||||
glob ["**/*"] [] b $ \fn -> flip runContT pure $ callCC \next -> do
|
||||
sz <- liftIO $ getFileSize fn
|
||||
-- when e2 $ mv inRefs (srcDir </> "refs")
|
||||
|
||||
when (sz >= 1024^3 ) do
|
||||
err $ red "Block is too large; skipping" <+> pretty fn
|
||||
next True
|
||||
-- pure srcDir
|
||||
|
||||
when (sz >= 1024^2 ) do
|
||||
warn $ yellow "Block is too large; but okay" <+> pretty fn
|
||||
-- tmp <- ContT $ Temp.withTempDirectory migrateDir "run"
|
||||
|
||||
let hs = nameToHash fn
|
||||
-- info $ "create dir" <+> pretty tmp
|
||||
|
||||
bs <- liftIO $ BS.copy <$> BS.readFile fn
|
||||
let h = HashRef $ hashObject @HbSync bs
|
||||
-- let blkz = source </> "blocks"
|
||||
-- let refz = source </> "refs"
|
||||
|
||||
unless ( h == hs ) do
|
||||
err $ red "Hash doesn't match content" <+> pretty fn
|
||||
next True
|
||||
-- let b = tmp </> "blocks"
|
||||
-- let r = tmp </> "refs"
|
||||
|
||||
placed <- liftIO $ ncqStoragePutBlock ncq (LBS.fromStrict bs)
|
||||
-- info $ "create directory links"
|
||||
-- info $ pretty blkz <+> pretty b
|
||||
-- liftIO $ createDirectoryLink blkz b
|
||||
|
||||
flush <- atomically do
|
||||
n <- readTVar cnt
|
||||
if n > 1000 then do
|
||||
writeTVar cnt 0
|
||||
pure True
|
||||
else do
|
||||
modifyTVar cnt succ
|
||||
pure False
|
||||
-- info $ pretty refz <+> pretty r
|
||||
-- liftIO $ createDirectoryLink refz r
|
||||
|
||||
unless ( placed == Just hs ) do
|
||||
err $ red "NCQ write error" <+> pretty fn
|
||||
next True
|
||||
-- ncq <- ContT $ withNCQ id ncqDir
|
||||
|
||||
when flush do
|
||||
liftIO (ncqStorageFlush ncq)
|
||||
-- let nameToHash fn =
|
||||
-- fromString @HashRef $ mconcat $ reverse $ take 2 $ reverse $ splitDirectories fn
|
||||
|
||||
for_ placed $ \hx -> atomically do
|
||||
writeTQueue checkQ (Just hx)
|
||||
modifyTVar checkN succ
|
||||
-- let hashToPath ha = do
|
||||
-- let (p,r) = splitAt 1 (show $ pretty ha)
|
||||
-- p </> r
|
||||
|
||||
info $ green "ok" <+> "B" <+> fill 44 (pretty placed) <+> pretty sz
|
||||
-- checkQ <- newTQueueIO
|
||||
-- checkN <- newTVarIO 0
|
||||
|
||||
pure True
|
||||
-- errors <- newTVarIO 0
|
||||
|
||||
unless norefs do
|
||||
glob ["**/*"] [] r $ \fn -> flip runContT pure $ callCC \next -> do
|
||||
-- rmp <- liftIO $ async $ fix \next -> do
|
||||
-- atomically (readTQueue checkQ) >>= \case
|
||||
-- Nothing -> none
|
||||
-- Just what -> do
|
||||
|
||||
let ref = nameToHash fn
|
||||
-- toWipe <- ncqLocate ncq what >>= \case
|
||||
-- Just (InCurrent{}) -> do
|
||||
-- atomically $ modifyTVar checkN pred
|
||||
-- pure True
|
||||
|
||||
ncqRef <- liftIO $ ncqStorageGetRef ncq ref
|
||||
-- Just (InFossil{}) -> do
|
||||
-- atomically $ modifyTVar checkN pred
|
||||
-- pure True
|
||||
|
||||
when (isJust ncqRef) do
|
||||
info $ yellow "keep" <+> "R" <+> pretty ref
|
||||
next True
|
||||
-- Just (InWriteQueue{}) -> do
|
||||
-- atomically $ unGetTQueue checkQ (Just what)
|
||||
-- pure False
|
||||
|
||||
refTo <- liftIO (readFile fn)
|
||||
<&> coerce @_ @HashRef . fromString @(Hash HbSync)
|
||||
-- Nothing -> do
|
||||
-- atomically $ modifyTVar errors succ
|
||||
-- pure False
|
||||
|
||||
here <- liftIO (ncqLocate ncq refTo)
|
||||
-- when toWipe do
|
||||
-- let path = b </> hashToPath what
|
||||
-- info $ yellow "d" <+> pretty what
|
||||
|
||||
if isJust here then do
|
||||
liftIO $ ncqStorageSetRef ncq ref refTo
|
||||
info $ green "ok" <+> "R" <+> pretty ref <+> pretty refTo
|
||||
else do
|
||||
warn $ red "Missed block for ref" <+> pretty ref <+> pretty refTo
|
||||
-- unless dry do
|
||||
-- rm path
|
||||
|
||||
pure True
|
||||
-- next
|
||||
|
||||
liftIO $ ncqIndexRightNow ncq
|
||||
-- cnt <- newTVarIO 0
|
||||
|
||||
info $ "check migration / wait to complete"
|
||||
-- glob ["**/*"] [] b $ \fn -> flip runContT pure $ callCC \next -> do
|
||||
-- sz <- liftIO $ getFileSize fn
|
||||
|
||||
atomically $ writeTQueue checkQ Nothing
|
||||
-- when (sz >= 1024^3 ) do
|
||||
-- err $ red "Block is too large; skipping" <+> pretty fn
|
||||
-- next True
|
||||
|
||||
wait rmp
|
||||
-- when (sz >= 1024^2 ) do
|
||||
-- warn $ yellow "Block is too large; but okay" <+> pretty fn
|
||||
|
||||
num <- readTVarIO checkN
|
||||
-- let hs = nameToHash fn
|
||||
|
||||
when (num == 0) $ exit ()
|
||||
-- bs <- liftIO $ BS.copy <$> BS.readFile fn
|
||||
-- let h = HashRef $ hashObject @HbSync bs
|
||||
|
||||
ee <- readTVarIO errors
|
||||
rest <- readTVarIO checkN
|
||||
-- unless ( h == hs ) do
|
||||
-- err $ red "Hash doesn't match content" <+> pretty fn
|
||||
-- next True
|
||||
|
||||
liftIO $ hPutDoc stdout $ "errors" <+> pretty ee <+> "leftovers" <+> pretty rest
|
||||
-- placed <- liftIO $ ncqStoragePutBlock ncq (LBS.fromStrict bs)
|
||||
|
||||
liftIO do
|
||||
if ee == 0 && rest == 0 then do
|
||||
-- flush <- atomically do
|
||||
-- n <- readTVar cnt
|
||||
-- if n > 1000 then do
|
||||
-- writeTVar cnt 0
|
||||
-- pure True
|
||||
-- else do
|
||||
-- modifyTVar cnt succ
|
||||
-- pure False
|
||||
|
||||
unless dry do
|
||||
rm migrateDir
|
||||
-- unless ( placed == Just hs ) do
|
||||
-- err $ red "NCQ write error" <+> pretty fn
|
||||
-- next True
|
||||
|
||||
exitSuccess
|
||||
-- when flush do
|
||||
-- liftIO (ncqStorageFlush ncq)
|
||||
|
||||
else
|
||||
exitFailure
|
||||
-- for_ placed $ \hx -> atomically do
|
||||
-- writeTQueue checkQ (Just hx)
|
||||
-- modifyTVar checkN succ
|
||||
|
||||
-- info $ green "ok" <+> "B" <+> fill 44 (pretty placed) <+> pretty sz
|
||||
|
||||
-- pure True
|
||||
|
||||
-- unless norefs do
|
||||
-- glob ["**/*"] [] r $ \fn -> flip runContT pure $ callCC \next -> do
|
||||
|
||||
-- let ref = nameToHash fn
|
||||
|
||||
-- ncqRef <- liftIO $ ncqStorageGetRef ncq ref
|
||||
|
||||
-- when (isJust ncqRef) do
|
||||
-- info $ yellow "keep" <+> "R" <+> pretty ref
|
||||
-- next True
|
||||
|
||||
-- refTo <- liftIO (readFile fn)
|
||||
-- <&> coerce @_ @HashRef . fromString @(Hash HbSync)
|
||||
|
||||
-- here <- liftIO (ncqLocate ncq refTo)
|
||||
|
||||
-- if isJust here then do
|
||||
-- liftIO $ ncqStorageSetRef ncq ref refTo
|
||||
-- info $ green "ok" <+> "R" <+> pretty ref <+> pretty refTo
|
||||
-- else do
|
||||
-- warn $ red "Missed block for ref" <+> pretty ref <+> pretty refTo
|
||||
|
||||
-- pure True
|
||||
|
||||
-- liftIO $ ncqIndexRightNow ncq
|
||||
|
||||
-- info $ "check migration / wait to complete"
|
||||
|
||||
-- atomically $ writeTQueue checkQ Nothing
|
||||
|
||||
-- wait rmp
|
||||
|
||||
-- num <- readTVarIO checkN
|
||||
|
||||
-- when (num == 0) $ exit ()
|
||||
|
||||
-- ee <- readTVarIO errors
|
||||
-- rest <- readTVarIO checkN
|
||||
|
||||
-- liftIO $ hPutDoc stdout $ "errors" <+> pretty ee <+> "leftovers" <+> pretty rest
|
||||
|
||||
-- liftIO do
|
||||
-- if ee == 0 && rest == 0 then do
|
||||
|
||||
-- unless dry do
|
||||
-- rm migrateDir
|
||||
|
||||
-- exitSuccess
|
||||
|
||||
-- else
|
||||
-- exitFailure
|
||||
|
||||
|
|
|
|||
|
|
@ -39,6 +39,49 @@ data PeerTcpProbeWaitKey
|
|||
data PeerUseHttpDownload
|
||||
data PeerBrainsDBPath
|
||||
|
||||
data PeerListenKey
|
||||
data PeerKeyFileKey
|
||||
data PeerStorageKey
|
||||
data PeerDebugKey
|
||||
data PeerTraceKey
|
||||
data PeerTrace1Key
|
||||
data PeerProxyFetchKey
|
||||
data PeerTcpSOCKS5
|
||||
data PeerDownloadThreadKey
|
||||
|
||||
|
||||
instance HasCfgKey PeerDebugKey a where
|
||||
key = "debug"
|
||||
|
||||
instance HasCfgKey PeerTraceKey a where
|
||||
key = "trace"
|
||||
|
||||
instance HasCfgKey PeerTrace1Key a where
|
||||
key = "trace1"
|
||||
|
||||
instance HasCfgKey PeerListenKey (Maybe String) where
|
||||
key = "listen"
|
||||
|
||||
instance HasCfgKey PeerKeyFileKey (Maybe String) where
|
||||
key = "key"
|
||||
|
||||
instance HasCfgKey PeerStorageKey (Maybe String) where
|
||||
key = "storage"
|
||||
|
||||
instance HasCfgKey PeerProxyFetchKey (Set String) where
|
||||
key = "proxy-fetch-for"
|
||||
|
||||
-- NOTE: socks5-auth
|
||||
-- Network.Simple.TCP does not support
|
||||
-- SOCKS5 authentification
|
||||
instance HasCfgKey PeerTcpSOCKS5 (Maybe String) where
|
||||
key = "tcp.socks5"
|
||||
|
||||
instance HasCfgKey PeerDownloadThreadKey (Maybe Int) where
|
||||
key = "download-threads"
|
||||
|
||||
|
||||
|
||||
newtype PeerHttpPort = PeerHttpPort (Maybe Integer)
|
||||
deriving newtype (Pretty)
|
||||
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ import HBS2.Net.Proto.Notify
|
|||
import HBS2.Peer.Proto.Mailbox
|
||||
import HBS2.OrDie
|
||||
import HBS2.Storage.Simple
|
||||
import HBS2.Storage.NCQ
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.Operations.Missed
|
||||
import HBS2.Data.Detect
|
||||
|
||||
|
|
@ -155,46 +155,6 @@ defStorageThreads :: Integral a => a
|
|||
defStorageThreads = 8
|
||||
|
||||
|
||||
data PeerListenKey
|
||||
data PeerKeyFileKey
|
||||
data PeerStorageKey
|
||||
data PeerDebugKey
|
||||
data PeerTraceKey
|
||||
data PeerTrace1Key
|
||||
data PeerProxyFetchKey
|
||||
data PeerTcpSOCKS5
|
||||
data PeerDownloadThreadKey
|
||||
|
||||
|
||||
instance HasCfgKey PeerDebugKey a where
|
||||
key = "debug"
|
||||
|
||||
instance HasCfgKey PeerTraceKey a where
|
||||
key = "trace"
|
||||
|
||||
instance HasCfgKey PeerTrace1Key a where
|
||||
key = "trace1"
|
||||
|
||||
instance HasCfgKey PeerListenKey (Maybe String) where
|
||||
key = "listen"
|
||||
|
||||
instance HasCfgKey PeerKeyFileKey (Maybe String) where
|
||||
key = "key"
|
||||
|
||||
instance HasCfgKey PeerStorageKey (Maybe String) where
|
||||
key = "storage"
|
||||
|
||||
instance HasCfgKey PeerProxyFetchKey (Set String) where
|
||||
key = "proxy-fetch-for"
|
||||
|
||||
-- NOTE: socks5-auth
|
||||
-- Network.Simple.TCP does not support
|
||||
-- SOCKS5 authentification
|
||||
instance HasCfgKey PeerTcpSOCKS5 (Maybe String) where
|
||||
key = "tcp.socks5"
|
||||
|
||||
instance HasCfgKey PeerDownloadThreadKey (Maybe Int) where
|
||||
key = "download-threads"
|
||||
|
||||
data PeerOpts =
|
||||
PeerOpts
|
||||
|
|
@ -825,7 +785,7 @@ runPeer opts = respawnOnError opts $ flip runContT pure do
|
|||
|
||||
let pref = fromMaybe xdg (view storage opts <|> storConf)
|
||||
|
||||
let ncqPath = coerce pref </> "ncq"
|
||||
let ncqPath = coerce pref </> "ncq3"
|
||||
|
||||
debug $ "storage prefix:" <+> pretty ncqPath
|
||||
|
||||
|
|
@ -864,7 +824,7 @@ runPeer opts = respawnOnError opts $ flip runContT pure do
|
|||
|
||||
-- error "STOP"
|
||||
|
||||
s <- lift $ ncqStorageOpen ncqPath
|
||||
s <- ContT $ ncqWithStorage ncqPath
|
||||
|
||||
-- simpleStorageInit @HbSync (Just pref)
|
||||
let blk = liftIO . hasBlock s
|
||||
|
|
@ -1435,6 +1395,7 @@ checkMigration prefix = flip runContT pure $ callCC \exit -> do
|
|||
already <- Sy.doesDirectoryExist migration
|
||||
|
||||
when (L.null blocks && not already) do
|
||||
checkNCQ1
|
||||
exit ()
|
||||
|
||||
let reason = if already then
|
||||
|
|
@ -1451,3 +1412,15 @@ checkMigration prefix = flip runContT pure $ callCC \exit -> do
|
|||
|
||||
|
||||
|
||||
where
|
||||
checkNCQ1 = do
|
||||
let ncq1Dir = prefix </> "ncq"
|
||||
ncq1Here <- Sy.doesDirectoryExist ncq1Dir
|
||||
when ncq1Here do
|
||||
notice $ yellow "found NCQv1 storage"
|
||||
notice $ red "Run" <+> "hbs2-peer migrate" <+> pretty prefix
|
||||
<> line
|
||||
<> "to migrate the storage to a new version"
|
||||
notice $ "You may also: backup" <+> pretty ncq1Dir <+> "or move it or remove permanently"
|
||||
liftIO exitFailure
|
||||
|
||||
|
|
|
|||
|
|
@ -189,6 +189,7 @@ library
|
|||
HBS2.Peer.RPC.Internal.Storage
|
||||
HBS2.Peer.RPC.Internal.Types
|
||||
HBS2.Peer.CLI.Detect
|
||||
HBS2.Peer.NCQ3.Migrate.NCQ
|
||||
|
||||
other-modules:
|
||||
-- HBS2.System.Logger.Simple
|
||||
|
|
|
|||
|
|
@ -0,0 +1,146 @@
|
|||
module HBS2.Peer.NCQ3.Migrate.NCQ where
|
||||
|
||||
|
||||
import HBS2.Base58
|
||||
import HBS2.Data.Types.Refs
|
||||
import HBS2.Hash
|
||||
import HBS2.Net.Auth.Schema()
|
||||
import HBS2.Peer.CLI.Detect
|
||||
import HBS2.Peer.RPC.API.LWWRef
|
||||
import HBS2.Peer.RPC.API.Peer
|
||||
import HBS2.Peer.RPC.API.RefLog
|
||||
import HBS2.Peer.RPC.API.Storage
|
||||
import HBS2.Peer.RPC.Client
|
||||
import HBS2.Peer.RPC.Client.Unix
|
||||
import HBS2.Storage
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.NCQ3.Internal.Files
|
||||
import HBS2.Storage.NCQ3.Internal.Fossil
|
||||
import HBS2.Storage.NCQ3.Internal.Index
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||
import HBS2.Peer.Proto.RefLog
|
||||
import HBS2.Peer.Proto.RefChan
|
||||
import HBS2.Peer.Proto.LWWRef
|
||||
|
||||
import Data.Config.Suckless.System
|
||||
import Data.Config.Suckless.Script
|
||||
|
||||
import Data.ByteString.Lazy qualified as LBS
|
||||
import Data.ByteString qualified as BS
|
||||
import Control.Monad.Trans.Cont
|
||||
import Control.Monad.Trans.Maybe
|
||||
import Data.HashSet (HashSet)
|
||||
import Data.HashSet qualified as HS
|
||||
import Data.HashMap.Strict qualified as HM
|
||||
import Data.HashMap.Strict (HashMap)
|
||||
import Data.Either
|
||||
import Data.List qualified as L
|
||||
import System.Directory (getModificationTime)
|
||||
import UnliftIO.IO
|
||||
|
||||
import Streaming.Prelude qualified as S
|
||||
|
||||
{- HLINT ignore "Functor law"-}
|
||||
|
||||
data WrapRef = forall a . ( Hashed HbSync a
|
||||
, Pretty a
|
||||
, RefMetaData a
|
||||
)
|
||||
=> WrapRef a
|
||||
|
||||
instance RefMetaData WrapRef where
|
||||
refMetaData (WrapRef a) = refMetaData a
|
||||
|
||||
instance Pretty WrapRef where
|
||||
pretty (WrapRef wtf) = pretty wtf
|
||||
|
||||
instance Hashed HbSync WrapRef where
|
||||
hashObject (WrapRef a) = hashObject a
|
||||
|
||||
legacyHashRefNCQ1 :: HashRef -> HashRef -> HashRef
|
||||
legacyHashRefNCQ1 salt h = HashRef (hashObject (coerce @_ @ByteString h <> coerce salt))
|
||||
|
||||
legacyHashRefFor :: Hashed HbSync p => HashRef -> p -> HashRef
|
||||
legacyHashRefFor salt x = legacyHashRefNCQ1 salt (HashRef $ hashObject @HbSync $ x)
|
||||
|
||||
migrateNCQ1 :: MonadUnliftIO m
|
||||
=> ( Doc AnsiStyle -> IO () )
|
||||
-> [WrapRef]
|
||||
-> FilePath
|
||||
-> FilePath
|
||||
-> m ()
|
||||
migrateNCQ1 logger refs src b = flip runContT pure do
|
||||
debug "migrate NCQ to NCQ3"
|
||||
|
||||
let a0 = src </> "0"
|
||||
a0here <- doesPathExist a0
|
||||
|
||||
let a = if a0here then a0 else src
|
||||
|
||||
ncq <- ContT $ ncqWithStorage b
|
||||
|
||||
let sto = AnyStorage ncq
|
||||
|
||||
let mlog = ncqGetFileName ncq "migrate-ncq1.log"
|
||||
|
||||
info $ "set SOURCE dir" <+> pretty a
|
||||
|
||||
ncqSalt <- liftIO (try @_ @IOException (BS.readFile (src </> ".seed")))
|
||||
<&> fromRight mempty
|
||||
<&> HashRef . hashObject
|
||||
|
||||
notice $ "ref. salt" <+> pretty ncqSalt
|
||||
|
||||
files <- do
|
||||
fs <- dirFiles a <&> L.filter (\x -> takeExtension x == ".data")
|
||||
mtimes <- liftIO $ mapM getModificationTime fs
|
||||
pure [ f | (f,_) <- L.sortOn snd (zip fs mtimes) ]
|
||||
|
||||
touch mlog
|
||||
|
||||
processed <- HS.fromList . lines <$> liftIO (readFile mlog)
|
||||
|
||||
let refz = HM.fromList [ (legacyHashRefFor ncqSalt r, WrapRef r) | WrapRef r <- refs ]
|
||||
|
||||
rv <- newTVarIO ( mempty :: HashMap HashRef (WrapRef, HashRef) )
|
||||
nblk <- newTVarIO 0
|
||||
|
||||
for_ files $ \f -> void $ runMaybeT do
|
||||
|
||||
guard ( not $ HS.member f processed )
|
||||
|
||||
debug $ "processing" <+> pretty f
|
||||
|
||||
ncqStorageScanDataFile0 f $ \offset w key s -> void $ runMaybeT do
|
||||
(t, bs) <- toMPlus (ncqEntryUnwrapValue s)
|
||||
|
||||
case t of
|
||||
B -> do
|
||||
let h0 = hashObject @HbSync bs
|
||||
if h0 /= coerce key then do
|
||||
err $ red "damaged block" <+> pretty key <+> "skip"
|
||||
else do
|
||||
void $ putBlock sto (LBS.fromStrict bs)
|
||||
atomically $ modifyTVar nblk succ
|
||||
|
||||
R -> do
|
||||
case HM.lookup key refz of
|
||||
Nothing -> none
|
||||
Just r -> do
|
||||
atomically $ modifyTVar rv (HM.insert (HashRef $ hashObject r) (r, HashRef (coerce bs)))
|
||||
|
||||
_ -> none
|
||||
|
||||
|
||||
liftIO $ appendFile mlog (f <> "\n")
|
||||
|
||||
nb <- readTVarIO nblk
|
||||
liftIO $ logger $ "moved" <+> pretty nb <+> "blocks"
|
||||
|
||||
foundRefs <- readTVarIO rv <&> HM.toList
|
||||
for_ foundRefs $ \(_, (r, v)) -> do
|
||||
blk <- hasBlock sto (coerce v)
|
||||
when (isJust blk) do
|
||||
liftIO $ logger $ green "update ref" <+> pretty r <+> pretty v
|
||||
updateRef sto r (coerce v)
|
||||
|
||||
|
|
@ -18,7 +18,7 @@ module HBS2.Storage.NCQ3
|
|||
|
||||
import HBS2.Storage.NCQ3.Internal.Types as Exported
|
||||
import HBS2.Storage.NCQ3.Internal.Class as Exported
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude as Exported
|
||||
-- import HBS2.Storage.NCQ3.Internal.Prelude as Exported
|
||||
import HBS2.Storage.NCQ3.Internal
|
||||
import HBS2.Storage.NCQ3.Internal.Run
|
||||
import HBS2.Storage.NCQ3.Internal.State
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import HBS2.Storage
|
|||
import HBS2.Storage.Simple
|
||||
import HBS2.Storage.Operations.ByteString
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||
import HBS2.Storage.NCQ3.Internal.Files
|
||||
import HBS2.Storage.NCQ3.Internal.Index
|
||||
import HBS2.Storage.NCQ3.Internal.Fossil
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import HBS2.Storage
|
|||
import HBS2.Storage.Simple
|
||||
import HBS2.Storage.Operations.ByteString
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||
import HBS2.Storage.NCQ3.Internal.Files
|
||||
import HBS2.Storage.NCQ3.Internal.Index
|
||||
import HBS2.Storage.NCQ3.Internal.Fossil
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import HBS2.Storage
|
|||
import HBS2.Storage.Simple
|
||||
import HBS2.Storage.Operations.ByteString
|
||||
import HBS2.Storage.NCQ3
|
||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||
import HBS2.Storage.NCQ3.Internal.Files
|
||||
import HBS2.Storage.NCQ3.Internal.Index
|
||||
import HBS2.Storage.NCQ3.Internal.Fossil
|
||||
|
|
|
|||
Loading…
Reference in New Issue