bypass reader when server is in backup mode

hstreamdb · Jun 26, 2024 · f8f9c9f · f8f9c9f
1 parent 390b6a2
commit f8f9c9f
Showing 1 changed file with 17 additions and 7 deletions.
diff --git a/hstream/src/HStream/Server/Core/ShardReader.hs b/hstream/src/HStream/Server/Core/ShardReader.hs
@@ -54,7 +54,7 @@ import           HStream.Server.Types        (BiStreamReader (..),
                                               BiStreamReaderSender,
                                               ServerContext (..),
                                               ServerInternalOffset,
-                                              ShardReader (..),
+                                              ServerMode (..), ShardReader (..),
                                               StreamReader (..), ToOffset (..),
                                               getLogLSN, mkShardReader,
                                               mkStreamReader)
@@ -143,12 +143,22 @@ readShard ServerContext{..} API.ReadShardRequest{..} = do
    readRecords r@ShardReader{..} = do
      let cStreamName = textToCBytes targetStream
      !read_start <- getPOSIXTime
-     records <- readProcessGap r (fromIntegral readShardRequestMaxRecords)
-     Stats.serverHistogramAdd scStatsHolder Stats.SHL_ReadLatency =<< msecSince read_start
-     Stats.stream_stat_add_read_in_bytes scStatsHolder cStreamName (fromIntegral . sum $ map (BS.length . S.recordPayload) records)
-     Stats.stream_stat_add_read_in_batches scStatsHolder cStreamName (fromIntegral $ length records)
-     let (records', _) = filterRecords shardReaderStartTs shardReaderEndTs records
-     receivedRecordsVecs <- forM records' decodeRecordBatch
+     -- records <- readProcessGap r (fromIntegral readShardRequestMaxRecords)
+     -- Stats.serverHistogramAdd scStatsHolder Stats.SHL_ReadLatency =<< msecSince read_start
+     -- Stats.stream_stat_add_read_in_bytes scStatsHolder cStreamName (fromIntegral . sum $ map (BS.length . S.recordPayload) records)
+     -- Stats.stream_stat_add_read_in_batches scStatsHolder cStreamName (fromIntegral $ length records)
+     -- let (records', _) = filterRecords shardReaderStartTs shardReaderEndTs records
+     -- receivedRecordsVecs <- forM records' decodeRecordBatch
+     state <- readIORef serverState
+     receivedRecordsVecs <- case state of
+       ServerNormal -> do
+         records <- readProcessGap r (fromIntegral readShardRequestMaxRecords)
+         Stats.serverHistogramAdd scStatsHolder Stats.SHL_ReadLatency =<< msecSince read_start
+         Stats.stream_stat_add_read_in_bytes scStatsHolder cStreamName (fromIntegral . sum $ map (BS.length . S.recordPayload) records)
+         Stats.stream_stat_add_read_in_batches scStatsHolder cStreamName (fromIntegral $ length records)
+         let (records', _) = filterRecords shardReaderStartTs shardReaderEndTs records
+         forM records' decodeRecordBatch
+       ServerBackup -> return []
      let res = V.fromList $ map (\(_, _, _, record) -> record) receivedRecordsVecs
      Log.debug $ "reader " <> Log.build readShardRequestReaderId
               <> " read " <> Log.build (V.length res) <> " batchRecords"