Debugging

Author: ChrisPenner

src/Share/Postgres.hs

@@ -100,6 +100,7 @@ import Share.Utils.Logging qualified as Logging
 import Share.Web.App
 import Share.Web.Errors (ErrorID (..), SomeServerError, ToServerError (..), internalServerError, respondError, someServerError)
 import System.CPUTime (getCPUTime)
+import Unison.Debug qualified as UDebug
 
 data TransactionError e
   = Unrecoverable SomeServerError
@@ -333,7 +334,9 @@ instance QueryA (Transaction e) where
   statement q s = do
     transactionStatement q s
 
-  unrecoverableError e = Transaction (pure (Left (Unrecoverable (someServerError e))))
+  unrecoverableError e = do
+    UDebug.debugM UDebug.Temp "Unrecoverable error in transaction: " e
+    Transaction (pure (Left (Unrecoverable (someServerError e))))
 
 instance QueryM (Transaction e) where
   transactionUnsafeIO io = Transaction (Right <$> liftIO io)
@@ -342,7 +345,9 @@ instance QueryA (Session e) where
   statement q s = do
     lift $ Session.statement q s
 
-  unrecoverableError e = throwError (Unrecoverable (someServerError e))
+  unrecoverableError e = do
+    UDebug.debugM UDebug.Temp "Unrecoverable error in transaction: " e
+    throwError (Unrecoverable (someServerError e))
 
 instance QueryM (Session e) where
   transactionUnsafeIO io = lift $ liftIO io
@@ -355,7 +360,8 @@ instance QueryA (Pipeline e) where
 instance (QueryM m) => QueryA (ReaderT e m) where
   statement q s = lift $ statement q s
 
-  unrecoverableError e = lift $ unrecoverableError e
+  unrecoverableError e = do
+    lift $ unrecoverableError e
 
 instance (QueryM m) => QueryM (ReaderT e m) where
   transactionUnsafeIO io = lift $ transactionUnsafeIO io

src/Share/Web/UCM/SyncV2/Impl.hs

@@ -8,7 +8,7 @@ import Control.Concurrent.STM qualified as STM
 import Control.Concurrent.STM.TBMQueue qualified as STM
 import Control.Monad.Except (ExceptT (ExceptT))
 import Control.Monad.Trans.Except (runExceptT)
-import Data.Conduit.Combinators qualified as Conduit
+import Data.List.NonEmpty qualified as NEL
 import Servant
 import Servant.Conduit (ConduitToSourceIO (..))
 import Servant.Types.SourceT (SourceT (..))
@@ -31,16 +31,20 @@ import Share.Web.Errors
 import Share.Web.UCM.Sync.HashJWT qualified as HashJWT
 import Share.Web.UCM.SyncV2.Queries qualified as SSQ
 import U.Codebase.Sqlite.Orphans ()
-import U.Codebase.Sqlite.TempEntity (TempEntity)
 import Unison.Debug qualified as Debug
-import Unison.Hash32 (Hash32)
 import Unison.Share.API.Hash (HashJWTClaims (..))
 import Unison.SyncV2.API qualified as SyncV2
-import Unison.SyncV2.Types (DownloadEntitiesChunk (..), EntityChunk (..), ErrorChunk (..))
+import Unison.SyncV2.Types (DownloadEntitiesChunk (..), EntityChunk (..), ErrorChunk (..), StreamInitInfo (..))
 import Unison.SyncV2.Types qualified as SyncV2
 import UnliftIO qualified
 import UnliftIO.Async qualified as Async
 
+batchSize :: Int32
+batchSize = 1000
+
+streamSettings :: StreamInitInfo
+streamSettings = StreamInitInfo {version = SyncV2.Version 1, entitySorting = SyncV2.Unsorted, numEntities = Nothing}
+
 server :: Maybe UserId -> SyncV2.Routes WebAppServer
 server mayUserId =
   SyncV2.Routes
@@ -59,7 +63,7 @@ parseBranchRef (SyncV2.BranchRef branchRef) =
     parseRelease = fmap Left . eitherToMaybe $ IDs.fromText @ProjectReleaseShortHand branchRef
 
 downloadEntitiesStreamImpl :: Maybe UserId -> SyncV2.DownloadEntitiesRequest -> WebApp (SourceIO SyncV2.DownloadEntitiesChunk)
-downloadEntitiesStreamImpl mayCallerUserId (SyncV2.DownloadEntitiesRequest {causalHash = causalHashJWT, branchRef, knownHashes= _todo}) = do
+downloadEntitiesStreamImpl mayCallerUserId (SyncV2.DownloadEntitiesRequest {causalHash = causalHashJWT, branchRef, knownHashes = _todo}) = do
   either emitErr id <$> runExceptT do
     addRequestTag "branch-ref" (SyncV2.unBranchRef branchRef)
     HashJWTClaims {hash = causalHash} <- lift (HashJWT.verifyHashJWT mayCallerUserId causalHashJWT >>= either respondError pure)
@@ -83,24 +87,35 @@ downloadEntitiesStreamImpl mayCallerUserId (SyncV2.DownloadEntitiesRequest {caus
           authZToken <- lift AuthZ.checkDownloadFromProjectBranchCodebase `whenLeftM` \_err -> throwError (SyncV2.DownloadEntitiesNoReadPermission branchRef)
           let codebaseLoc = Codebase.codebaseLocationForProjectBranchCodebase projectOwnerUserId contributorId
           pure $ Codebase.codebaseEnv authZToken codebaseLoc
-    q <- liftIO $ STM.newTBMQueueIO 10
+    q <- UnliftIO.atomically $ do
+      q <- STM.newTBMQueue 10
+      STM.writeTBMQueue q (NEL.singleton $ InitialC $ streamSettings)
+      pure q
     streamResults <- lift $ UnliftIO.toIO do
       Logging.logInfoText "Starting download entities stream"
       Codebase.runCodebaseTransaction codebase $ do
+        Debug.debugM Debug.Temp "Getting IDs for:" causalHash
         (_bhId, causalId) <- CausalQ.expectCausalIdsOf id (hash32ToCausalHash causalHash)
+        Debug.debugM Debug.Temp "Getting deps of" causalId
         cursor <- SSQ.allSerializedDependenciesOfCausalCursor causalId
-        Cursor.foldBatched cursor 1000 \batch -> do
-          PG.transactionUnsafeIO $ STM.atomically $ STM.writeTBMQueue q batch
+        Debug.debugLogM Debug.Temp "Got cursor"
+        Cursor.foldBatched cursor batchSize \batch -> do
+          Debug.debugLogM Debug.Temp "Emitting batch"
+          let entityChunkBatch = batch <&> \(entityCBOR, hash) -> EntityC (EntityChunk {hash, entityCBOR})
+          PG.transactionUnsafeIO $ STM.atomically $ STM.writeTBMQueue q entityChunkBatch
         PG.transactionUnsafeIO $ STM.atomically $ STM.closeTBMQueue q
-    pure $ sourceIOWithAsync streamResults $ conduitToSourceIO do
+    liftIO $ Async.async streamResults
+    -- pure $ sourceIOWithAsync streamResults $ conduitToSourceIO do
+    pure $ conduitToSourceIO do
       stream q
-        Conduit..| ( Conduit.iterM \case
-                       InitialC init -> Debug.debugM Debug.Temp "Initial " init
-                       EntityC ec -> Debug.debugM Debug.Temp "Chunk " ec
-                       ErrorC err -> Debug.debugM Debug.Temp "Error " err
-                   )
   where
-    stream :: STM.TBMQueue (NonEmpty (SyncV2.CBORBytes TempEntity, Hash32)) -> ConduitT () DownloadEntitiesChunk IO ()
+    -- Conduit..| ( Conduit.iterM \case
+    --                InitialC init -> Debug.debugM Debug.Temp "Initial " init
+    --                EntityC ec -> Debug.debugM Debug.Temp "Chunk " ec
+    --                ErrorC err -> Debug.debugM Debug.Temp "Error " err
+    --            )
+
+    stream :: STM.TBMQueue (NonEmpty DownloadEntitiesChunk) -> ConduitT () DownloadEntitiesChunk IO ()
     stream q = do
       let loop :: ConduitT () DownloadEntitiesChunk IO ()
           loop = do
@@ -111,18 +126,18 @@ downloadEntitiesStreamImpl mayCallerUserId (SyncV2.DownloadEntitiesRequest {caus
                 Debug.debugLogM Debug.Temp "Queue closed. finishing up!"
                 pure ()
               Just batch -> do
-                let chunks = batch <&> \(entityCBOR, hash) -> EntityC (EntityChunk {hash, entityCBOR})
-                Debug.debugLogM Debug.Temp $ "Emitting chunk of " <> show (length chunks) <> " entities"
-                yieldMany chunks
+                Debug.debugLogM Debug.Temp $ "Emitting chunk of " <> show (length batch) <> " entities"
+                yieldMany batch
                 loop
+
       loop
       Debug.debugLogM Debug.Temp "Done!"
 
     emitErr :: SyncV2.DownloadEntitiesError -> SourceIO SyncV2.DownloadEntitiesChunk
     emitErr err = SourceT.source [ErrorC (ErrorChunk err)]
 
 -- | Run an IO action in the background while streaming the results.
-sourceIOWithAsync :: IO a -> SourceIO r -> SourceIO r
-sourceIOWithAsync action (SourceT k) =
+_sourceIOWithAsync :: IO a -> SourceIO r -> SourceIO r
+_sourceIOWithAsync action (SourceT k) =
   SourceT \k' ->
     Async.withAsync action \_ -> k k'