address more PR comments

Author: mitchellwrosen

src/Share/BackgroundJobs/Diffs/ContributionDiffs.hs

@@ -37,16 +37,15 @@ worker scope = do
   unisonRuntime <- asks Env.sandboxedRuntime
   let makeRuntime :: Codebase.CodebaseEnv -> IO (Codebase.CodebaseRuntime IO)
       makeRuntime codebase = do
-        runtime <- Codebase.codebaseRuntime' unisonRuntime codebase
+        runtime <- Codebase.codebaseRuntimeTransaction unisonRuntime codebase
         pure (badUnliftCodebaseRuntime runtime)
   newWorker scope "diffs:contributions" $ forever do
     Notif.waitOnChannel Notif.ContributionDiffChannel (maxPollingIntervalSeconds * 1000000)
     processDiffs authZReceipt makeRuntime
 
--- Process diffs until we run out of them. We claim a diff in a transaction, commit it, then proceed to compute the
--- diff. There's therefore a chance we claim a diff and fail to compute it (due to e.g. server restart). The current
--- solution to these "at most once" semantics is to simply re-enqueue a diff job if necessary; e.g. in the view diff
--- endpoint handler.
+-- Process diffs until we run out of them. We claim a diff in a transaction and compute the diff in the same
+-- transaction, with a row lock on the contribution id (which is skipped by other workers). There's therefore no chance
+-- that we claim a diff but fail to write the result of computing that diff back to the database.
 processDiffs :: AuthZ.AuthZReceipt -> (Codebase.CodebaseEnv -> IO (Codebase.CodebaseRuntime IO)) -> Background ()
 processDiffs authZReceipt makeRuntime = do
   let loop :: Background ()

src/Share/Codebase.hs

@@ -17,7 +17,7 @@ module Share.Codebase
     CodebaseRuntime (..),
     codebaseEnv,
     codebaseRuntime,
-    codebaseRuntime',
+    codebaseRuntimeTransaction,
     badAskUnliftCodebaseRuntime,
     codebaseForProjectBranch,
     codebaseLocationForUserCodebase,
@@ -189,16 +189,16 @@ codebaseEnv !_authZReceipt codebaseLoc = do
 codebaseRuntime :: (MonadReader (Env.Env x) m, MonadUnliftIO m) => CodebaseEnv -> m (CodebaseRuntime IO)
 codebaseRuntime codebase = do
   unisonRuntime <- asks Env.sandboxedRuntime
-  rt <- liftIO (codebaseRuntime' unisonRuntime codebase)
+  rt <- liftIO (codebaseRuntimeTransaction unisonRuntime codebase)
   unlift <- badAskUnliftCodebaseRuntime
   pure (unlift rt)
 
 -- | Ideally, we'd use this – a runtime with lookup actions in transaction, not IO. But that will require refactoring to
 -- the runtime interface in ucm, so we can't use it for now. That's bad: we end up unsafely running separate
 -- transactions for inner calls to 'codeLookup' / 'cachedEvalResult', which can lead to deadlock due to a starved
 -- connection pool.
-codebaseRuntime' :: Runtime Symbol -> CodebaseEnv -> IO (CodebaseRuntime (PG.Transaction e))
-codebaseRuntime' unisonRuntime CodebaseEnv {codebaseOwner} = do
+codebaseRuntimeTransaction :: Runtime Symbol -> CodebaseEnv -> IO (CodebaseRuntime (PG.Transaction e))
+codebaseRuntimeTransaction unisonRuntime CodebaseEnv {codebaseOwner} = do
   cacheVar <- newTVarIO (CodeLookupCache mempty mempty)
   pure
     CodebaseRuntime
@@ -207,6 +207,9 @@ codebaseRuntime' unisonRuntime CodebaseEnv {codebaseOwner} = do
         unisonRuntime
       }
 
+-- Why bad: see above comment on `codebaseRuntimeTransaction`. We don't want to use a `CodebaseRuntime IO`, because it
+-- will run every lookup in a separate transaction. But we can't use a `CodebaseRuntime Transaction` because we call
+-- back into UCM library code that expects a `CodebaseRuntime IO`.
 badAskUnliftCodebaseRuntime ::
   (MonadReader (Env.Env x) m, MonadUnliftIO m) =>
   m (CodebaseRuntime (PG.Transaction Void) -> CodebaseRuntime IO)

src/Share/NamespaceDiffs.hs

@@ -817,10 +817,10 @@ computeThreeWayNamespaceDiff codebaseEnvs2 branchHashIds3 nameLookupReceipts3 =
           let termReferenceIds = Map.mapMaybe Referent.toTermReferenceId (BiMultimap.range termReferents)
           termIds <-
             PG.pFor termReferenceIds \refId ->
-              (refId,) <$> DefnsQ.pipelinedExpectTermId refId
+              (refId,) <$> DefnsQ.expectTermId refId
           v2Terms <-
             PG.pFor termIds \(refId, termId) ->
-              (refId,) <$> DefnsQ.pipelinedExpectTermById codebaseUser refId termId
+              (refId,) <$> DefnsQ.expectTermById codebaseUser refId termId
           v1Terms <-
             for v2Terms \(refId, (term, typ)) ->
               (refId,) <$> Codebase.convertTerm2to1 (Reference.idToHash refId) term typ
@@ -833,7 +833,7 @@ computeThreeWayNamespaceDiff codebaseEnvs2 branchHashIds3 nameLookupReceipts3 =
           let typeReferenceIds = Map.mapMaybe Reference.toId (BiMultimap.range typeReferences)
           typeIds <-
             PG.pFor typeReferenceIds \refId ->
-              (refId,) <$> DefnsQ.pipelinedExpectTypeComponentElementAndTypeId codebaseUser refId
+              (refId,) <$> DefnsQ.expectTypeComponentElementAndTypeId codebaseUser refId
           v1Decls <-
             PG.pFor typeIds \(refId, typeId) ->
               DefnsQ.loadDeclByTypeComponentElementAndTypeId typeId <&> \v2Decl ->

src/Share/Postgres.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE DefaultSignatures #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE ImpredicativeTypes #-}
 {-# LANGUAGE LiberalTypeSynonyms #-}
@@ -45,7 +46,6 @@ module Share.Postgres
     defaultIsolationLevel,
     pipelined,
     pEitherMap,
-    pUnrecoverableEitherMap,
     pFor,
     pFor_,
 
@@ -139,14 +139,6 @@ pEitherMap f (Pipeline p) =
       Right x -> mapLeft Err (f x)
       Left e -> Left e
 
--- | Like 'pEitherMap', but for throwing unrecoverable errors.
-pUnrecoverableEitherMap :: (Loggable e, Show e, ToServerError e) => (a -> Either e b) -> Pipeline e' a -> Pipeline e' b
-pUnrecoverableEitherMap f (Pipeline p) =
-  Pipeline $
-    p <&> \case
-      Right x -> mapLeft (Unrecoverable . someServerError) (f x)
-      Left e -> Left e
-
 pFor :: (Traversable f) => f a -> (a -> Pipeline e b) -> Transaction e (f b)
 pFor f p = pipelined $ for f p
 
@@ -346,6 +338,17 @@ class (Applicative m) => QueryA m where
   -- | Fail the transaction and whole request with an unrecoverable server error.
   unrecoverableError :: (HasCallStack, ToServerError e, Loggable e, Show e) => e -> m a
 
+  -- | Map an either-returning function over the result of an action; if it returns Left, throw an unrecoverable error.
+  -- This is a trivial combinator for any monad, hence the default signature, but it can be implemented by our
+  -- Pipeline applicative, too.
+  unrecoverableEitherMap :: (HasCallStack, Loggable e, Show e, ToServerError e) => (a -> Either e b) -> m a -> m b
+  default unrecoverableEitherMap :: (HasCallStack, Loggable e, Show e, ToServerError e, Monad m) => (a -> Either e b) -> m a -> m b
+  unrecoverableEitherMap f m = do
+    x <- m
+    case f x of
+      Right y -> pure y
+      Left e -> unrecoverableError e
+
 class (Monad m, QueryA m) => QueryM m where
   -- | Allow running IO actions in a transaction. These actions may be run multiple times if
   -- the transaction is retried.
@@ -376,6 +379,12 @@ instance QueryA (Pipeline e) where
 
   unrecoverableError e = Pipeline $ pure (Left (Unrecoverable (someServerError e)))
 
+  unrecoverableEitherMap f (Pipeline p) =
+    Pipeline $
+      p <&> \case
+        Right x -> mapLeft (Unrecoverable . someServerError) (f x)
+        Left e -> Left e
+
 instance (QueryM m) => QueryA (ReaderT e m) where
   statement q s = lift $ statement q s
 

src/Share/Postgres/Causal/Queries.hs

@@ -20,7 +20,6 @@ module Share.Postgres.Causal.Queries
     expectNamespaceStatsOf,
     expectNamespaceHashByCausalHash,
     HashQ.expectNamespaceIdsByCausalIdsOf,
-    HashQ.pipelinedExpectNamespaceIdsByCausalIdsOf,
     importAccessibleCausals,
     importCausalIntoCodebase,
     hashCausal,

src/Share/Postgres/Definitions/Queries.hs

@@ -3,8 +3,8 @@
 module Share.Postgres.Definitions.Queries
   ( loadTerm,
     expectTerm,
-    pipelinedExpectTermId,
-    pipelinedExpectTermById,
+    expectTermId,
+    expectTermById,
     saveTermComponent,
     saveEncodedTermComponent,
     loadTermComponent,
@@ -20,7 +20,7 @@ module Share.Postgres.Definitions.Queries
     loadDecl,
     expectDecl,
     loadDeclByTypeComponentElementAndTypeId,
-    pipelinedExpectTypeComponentElementAndTypeId,
+    expectTypeComponentElementAndTypeId,
     loadCachedEvalResult,
     saveCachedEvalResult,
     termReferencesByPrefix,
@@ -153,9 +153,9 @@ loadTermId (Reference.Id compHash (pgComponentIndex -> compIndex)) =
             AND term.component_index = #{compIndex}
       |]
 
-pipelinedExpectTermId :: TermReferenceId -> PG.Pipeline e TermId
-pipelinedExpectTermId refId =
-  pUnrecoverableEitherMap
+expectTermId :: QueryA m => TermReferenceId -> m TermId
+expectTermId refId =
+  unrecoverableEitherMap
     ( \case
         Nothing -> Left (expectedTermError refId)
         Just termId -> Right termId
@@ -310,9 +310,9 @@ loadTermById codebaseUser termId = do
     <$> loadTermComponentElementByTermId codebaseUser termId
     <*> termLocalReferences termId
 
-pipelinedExpectTermById :: UserId -> TermReferenceId -> TermId -> Pipeline e (V2.Term Symbol, V2.Type Symbol)
-pipelinedExpectTermById userId refId termId =
-  pUnrecoverableEitherMap
+expectTermById :: QueryA m => UserId -> TermReferenceId -> TermId -> m (V2.Term Symbol, V2.Type Symbol)
+expectTermById userId refId termId =
+  unrecoverableEitherMap
     ( \case
         Nothing -> Left (expectedTermError refId)
         Just term -> Right term
@@ -443,9 +443,9 @@ loadTypeComponentElementAndTypeId codebaseUser (Reference.Id compHash (pgCompone
             AND typ.component_index = #{compIndex}
       |]
 
-pipelinedExpectTypeComponentElementAndTypeId :: UserId -> TermReferenceId -> PG.Pipeline e (TypeComponentElement, TypeId)
-pipelinedExpectTypeComponentElementAndTypeId codebaseUser refId =
-  pUnrecoverableEitherMap
+expectTypeComponentElementAndTypeId :: QueryA m => UserId -> TermReferenceId -> m (TypeComponentElement, TypeId)
+expectTypeComponentElementAndTypeId codebaseUser refId =
+  unrecoverableEitherMap
     ( \case
         Nothing -> Left (expectedTypeError refId)
         Just decl -> Right decl

src/Share/Postgres/Hashes/Queries.hs

@@ -29,7 +29,6 @@ module Share.Postgres.Hashes.Queries
     loadCausalIdByHash,
     expectCausalIdByHash,
     expectNamespaceIdsByCausalIdsOf,
-    pipelinedExpectNamespaceIdsByCausalIdsOf,
     expectNamespaceHashesByNamespaceHashIdsOf,
     isComponentHashAllowedToBeMismatched,
     isCausalHashAllowedToBeMismatched,
@@ -305,31 +304,6 @@ expectNamespaceIdsByCausalIdsOf trav s = do
         then unrecoverableError . MissingExpectedEntity $ "expectNamespaceIdsByCausalIdsOf: Expected to get the same number of results as causal ids. " <> tShow causalIds
         else pure results
 
--- | Mitchell says: this could/should just have replaced 'expectNamespaceIdsByCausalIdsOf', but that function has
--- many callers, so having two temporarily eases the transition.
-pipelinedExpectNamespaceIdsByCausalIdsOf :: Traversal s t CausalId BranchHashId -> s -> Pipeline e t
-pipelinedExpectNamespaceIdsByCausalIdsOf trav s = do
-  s
-    & unsafePartsOf trav %%~ \causalIds ->
-      let causalIdsTable = ordered causalIds
-       in pUnrecoverableEitherMap
-            ( \results ->
-                if length results /= length causalIds
-                  then Left (MissingExpectedEntity $ "expectNamespaceIdsByCausalIdsOf: Expected to get the same number of results as causal ids. " <> tShow causalIds)
-                  else Right results
-            )
-            ( queryListCol @BranchHashId
-                [sql|
-                  WITH causal_ids(ord, causal_id) AS (
-                    SELECT ord, causal_id FROM ^{toTable causalIdsTable} as t(ord, causal_id)
-                  )
-                  SELECT c.namespace_hash_id
-                    FROM causal_ids cid
-                    JOIN causals c ON cid.causal_id = c.id
-                    ORDER BY cid.ord
-                  |]
-            )
-
 expectNamespaceHashesByNamespaceHashIdsOf :: (HasCallStack, QueryM m) => Traversal s t BranchHashId BranchHash -> s -> m t
 expectNamespaceHashesByNamespaceHashIdsOf trav s = do
   s

src/Share/Web/Share/Projects/Impl.hs

@@ -175,7 +175,7 @@ diffNamespacesEndpoint (AuthN.MaybeAuthedUserID callerUserId) userHandle project
     unisonRuntime <- asks Env.sandboxedRuntime
     let makeRuntime :: Codebase.CodebaseEnv -> IO (Codebase.CodebaseRuntime IO)
         makeRuntime codebase = do
-          runtime <- Codebase.codebaseRuntime' unisonRuntime codebase
+          runtime <- Codebase.codebaseRuntimeTransaction unisonRuntime codebase
           pure (badUnliftCodebaseRuntime runtime)
     diff <-
       PG.runTransaction do