mirror of
https://github.com/urbit/shrub.git
synced 2024-12-26 21:44:11 +03:00
Merge pull request #2236 from urbit/bs/fix-log-replay
king: Fix log replay bug.
This commit is contained in:
commit
8836513a7c
@ -131,13 +131,19 @@ resumed :: (HasStderrLogFunc e, HasPierConfig e, HasLogFunc e)
|
|||||||
=> Maybe Word64 -> Serf.Flags
|
=> Maybe Word64 -> Serf.Flags
|
||||||
-> RAcquire e (Serf e, EventLog, SerfState)
|
-> RAcquire e (Serf e, EventLog, SerfState)
|
||||||
resumed event flags = do
|
resumed event flags = do
|
||||||
|
rio $ logTrace "Resuming ship"
|
||||||
top <- view pierPathL
|
top <- view pierPathL
|
||||||
tap <- fmap (fromMaybe top) $ rio $ runMaybeT $ do
|
tap <- fmap (fromMaybe top) $ rio $ runMaybeT $ do
|
||||||
ev <- MaybeT (pure event)
|
ev <- MaybeT (pure event)
|
||||||
MaybeT (getSnapshot top ev)
|
MaybeT (getSnapshot top ev)
|
||||||
rio $ logTrace $ displayShow tap
|
|
||||||
|
rio $ logTrace $ display @Text ("pier: " <> pack top)
|
||||||
|
rio $ logTrace $ display @Text ("running serf in: " <> pack tap)
|
||||||
|
|
||||||
log <- Log.existing (top <> "/.urb/log")
|
log <- Log.existing (top <> "/.urb/log")
|
||||||
|
|
||||||
serf <- Serf.run (Serf.Config tap flags)
|
serf <- Serf.run (Serf.Config tap flags)
|
||||||
|
|
||||||
serfSt <- rio $ Serf.replay serf log event
|
serfSt <- rio $ Serf.replay serf log event
|
||||||
|
|
||||||
rio $ Serf.snapshot serf serfSt
|
rio $ Serf.snapshot serf serfSt
|
||||||
|
@ -270,7 +270,9 @@ cordText = T.strip . unCord
|
|||||||
--------------------------------------------------------------------------------
|
--------------------------------------------------------------------------------
|
||||||
|
|
||||||
snapshot :: HasLogFunc e => Serf e -> SerfState -> RIO e ()
|
snapshot :: HasLogFunc e => Serf e -> SerfState -> RIO e ()
|
||||||
snapshot serf ss = sendOrder serf $ OSave $ ssLastEv ss
|
snapshot serf ss = do
|
||||||
|
logTrace $ display ("Taking snapshot at event " <> tshow (ssLastEv ss))
|
||||||
|
sendOrder serf $ OSave $ ssLastEv ss
|
||||||
|
|
||||||
shutdown :: HasLogFunc e => Serf e -> Word8 -> RIO e ()
|
shutdown :: HasLogFunc e => Serf e -> Word8 -> RIO e ()
|
||||||
shutdown serf code = sendOrder serf (OExit code)
|
shutdown serf code = sendOrder serf (OExit code)
|
||||||
@ -298,12 +300,20 @@ recvPlea w = do
|
|||||||
-}
|
-}
|
||||||
handshake :: HasLogFunc e => Serf e -> LogIdentity -> RIO e SerfState
|
handshake :: HasLogFunc e => Serf e -> LogIdentity -> RIO e SerfState
|
||||||
handshake serf ident = do
|
handshake serf ident = do
|
||||||
|
logTrace "Serf Handshake"
|
||||||
|
|
||||||
ss@SerfState{..} <- recvPlea serf >>= \case
|
ss@SerfState{..} <- recvPlea serf >>= \case
|
||||||
PPlay e m -> pure $ SerfState e m
|
PPlay e m -> pure $ SerfState e m
|
||||||
x -> throwIO (InvalidInitialPlea x)
|
x -> throwIO (InvalidInitialPlea x)
|
||||||
|
|
||||||
|
logTrace $ display ("Handshake result: " <> tshow ss)
|
||||||
|
|
||||||
when (ssNextEv == 1) $ do
|
when (ssNextEv == 1) $ do
|
||||||
sendOrder serf (OBoot (lifecycleLen ident))
|
let ev = OBoot (lifecycleLen ident)
|
||||||
|
logTrace $ display ("No snapshot. Sending boot event: " <> tshow ev)
|
||||||
|
sendOrder serf ev
|
||||||
|
|
||||||
|
logTrace "Finished handshake"
|
||||||
|
|
||||||
pure ss
|
pure ss
|
||||||
|
|
||||||
@ -435,27 +445,43 @@ replayJobs serf lastEv = go Nothing
|
|||||||
go pb played
|
go pb played
|
||||||
|
|
||||||
updatePb ss = do
|
updatePb ss = do
|
||||||
let start = lastEv - (fromIntegral (ssNextEv ss))
|
let start = lastEv - fromIntegral (ssNextEv ss)
|
||||||
let msg = pack ("Replaying events #" ++ (show (ssNextEv ss)) ++
|
let msg = pack ( "Replaying events #" ++ (show (ssNextEv ss))
|
||||||
" to #" ++ (show lastEv))
|
<> " to #" ++ (show lastEv)
|
||||||
|
)
|
||||||
updateProgressBar start msg
|
updateProgressBar start msg
|
||||||
|
|
||||||
|
|
||||||
replay :: (HasStderrLogFunc e, HasLogFunc e)
|
replay :: (HasStderrLogFunc e, HasLogFunc e)
|
||||||
=> Serf e -> Log.EventLog -> Maybe Word64 -> RIO e SerfState
|
=> Serf e -> Log.EventLog -> Maybe Word64 -> RIO e SerfState
|
||||||
replay serf log last = do
|
replay serf log last = do
|
||||||
|
logTrace "Beginning event log replay"
|
||||||
|
|
||||||
|
last & \case
|
||||||
|
Nothing -> pure ()
|
||||||
|
Just lt -> logTrace $ display $
|
||||||
|
"User requested to replay up to event #" <> tshow lt
|
||||||
|
|
||||||
ss <- handshake serf (Log.identity log)
|
ss <- handshake serf (Log.identity log)
|
||||||
|
|
||||||
let numEvs = case last of
|
logLastEv :: Word64 <- fromIntegral <$> Log.lastEv log
|
||||||
Nothing -> ssNextEv ss
|
|
||||||
Just la -> la - (ssNextEv ss) + 1
|
|
||||||
|
|
||||||
lastEv <- last & maybe (Log.lastEv log) pure
|
let serfNextEv = ssNextEv ss
|
||||||
|
lastEventInSnap = serfNextEv - 1
|
||||||
|
|
||||||
runConduit $ Log.streamEvents log (ssNextEv ss)
|
logTrace $ display $ "Last event in event log is #" <> tshow logLastEv
|
||||||
|
|
||||||
|
let replayUpTo = fromMaybe logLastEv last
|
||||||
|
|
||||||
|
let numEvs :: Int = fromIntegral replayUpTo - fromIntegral lastEventInSnap
|
||||||
|
|
||||||
|
logTrace $ display $ "Replaying up to event #" <> tshow replayUpTo
|
||||||
|
logTrace $ display $ "Will replay " <> tshow numEvs <> " in total."
|
||||||
|
|
||||||
|
runConduit $ Log.streamEvents log serfNextEv
|
||||||
.| CC.take (fromIntegral numEvs)
|
.| CC.take (fromIntegral numEvs)
|
||||||
.| toJobs (Log.identity log) (ssNextEv ss)
|
.| toJobs (Log.identity log) serfNextEv
|
||||||
.| replayJobs serf (fromIntegral lastEv) ss
|
.| replayJobs serf (fromIntegral replayUpTo) ss
|
||||||
|
|
||||||
toJobs :: HasLogFunc e
|
toJobs :: HasLogFunc e
|
||||||
=> LogIdentity -> EventId -> ConduitT ByteString Job (RIO e) ()
|
=> LogIdentity -> EventId -> ConduitT ByteString Job (RIO e) ()
|
||||||
|
Loading…
Reference in New Issue
Block a user