Merge pull request #29 from unisoncomputing/cp/batched-diffs

Batched background diffs

Author: ChrisPenner

share-api.cabal

@@ -28,6 +28,8 @@ library
       Share.App
       Share.Backend
       Share.BackgroundJobs
+      Share.BackgroundJobs.Diffs.ContributionDiffs
+      Share.BackgroundJobs.Diffs.Queries
       Share.BackgroundJobs.Errors
       Share.BackgroundJobs.Monad
       Share.BackgroundJobs.Search.DefinitionSync

share-utils/package.yaml

@@ -42,6 +42,7 @@ default-extensions:
   - FlexibleContexts
   - FlexibleInstances
   - GeneralizedNewtypeDeriving
+  - InstanceSigs
   - LambdaCase
   - MultiParamTypeClasses
   - NamedFieldPuns

share-utils/share-utils.cabal

@@ -1,6 +1,6 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.36.0.
+-- This file has been generated from package.yaml by hpack version 0.37.0.
 --
 -- see: https://github.com/sol/hpack
 
@@ -25,6 +25,7 @@ source-repository head
 library
   exposed-modules:
       Share.Debug
+      Share.Utils.Aeson
       Share.Utils.Binary
       Share.Utils.Deployment
       Share.Utils.IDs
@@ -50,6 +51,7 @@ library
       FlexibleContexts
       FlexibleInstances
       GeneralizedNewtypeDeriving
+      InstanceSigs
       LambdaCase
       MultiParamTypeClasses
       NamedFieldPuns

share-utils/src/Share/Utils/Aeson.hs

@@ -0,0 +1,40 @@
+{-# LANGUAGE InstanceSigs #-}
+
+module Share.Utils.Aeson (MaybeEncoded (..), PreEncoded (..)) where
+
+import Data.Aeson (ToJSON (..))
+import Data.Aeson qualified as Aeson
+import Data.Aeson.Encoding qualified as Encoding
+import Data.Binary.Builder qualified as Builder
+import Data.ByteString.Lazy.Char8 qualified as BL
+import Data.Maybe (fromMaybe)
+import Data.Proxy (Proxy (..))
+import Data.Typeable (Typeable, typeRep)
+import GHC.Stack (HasCallStack)
+
+data MaybeEncoded a
+  = IsEncoded BL.ByteString
+  | NotEncoded a
+  deriving stock (Show, Eq, Ord, Functor, Foldable, Traversable)
+
+instance (Typeable a, ToJSON a) => ToJSON (MaybeEncoded a) where
+  toJSON :: (HasCallStack) => MaybeEncoded a -> Aeson.Value
+  toJSON (IsEncoded txt) = Aeson.toJSON (PreEncoded @a txt)
+  toJSON (NotEncoded a) = Aeson.toJSON a
+
+  toEncoding :: (HasCallStack) => MaybeEncoded a -> Aeson.Encoding
+  toEncoding (IsEncoded txt) = Aeson.toEncoding (PreEncoded @a txt)
+  toEncoding (NotEncoded a) = Aeson.toEncoding a
+
+newtype PreEncoded a = PreEncoded BL.ByteString
+  deriving stock (Show, Eq, Ord)
+
+instance (Typeable a) => ToJSON (PreEncoded a) where
+  toJSON :: (HasCallStack) => PreEncoded a -> Aeson.Value
+  toJSON (PreEncoded txt) =
+    -- It's regrettable we have to do this, but seemingly it's required when building values
+    -- with @@object [key .= val]@@ syntax.
+    fromMaybe (error $ "Invalid PreEncoded JSON for type: " <> show (typeRep (Proxy @a))) $ Aeson.decode txt
+
+  toEncoding :: (HasCallStack) => PreEncoded a -> Aeson.Encoding
+  toEncoding (PreEncoded txt) = Encoding.unsafeToEncoding . Builder.fromLazyByteString $ txt

share-utils/src/Share/Utils/URI.hs

@@ -22,12 +22,12 @@ import Data.Map qualified as Map
 import Data.Text (Text)
 import Data.Text qualified as Text
 import Data.Text.Encoding qualified as Text
-import Share.Utils.Show (tShow)
 import Hasql.Decoders qualified as Decoders
 import Hasql.Interpolate qualified as Hasql
 import Network.HTTP.Types (parseQuery, renderQuery)
 import Network.URI qualified as URI
 import Servant
+import Share.Utils.Show (tShow)
 
 -- | Helper type to provide additional instances for URIs.
 newtype URIParam = URIParam {unpackURI :: URI}

sql/2024-11-19-00-00_namespace_diffs.sql

@@ -0,0 +1,23 @@
+-- Adds tables for storing pre-computed namespace diffs
+
+CREATE TABLE namespace_diffs (
+    left_namespace_id INTEGER NOT NULL REFERENCES branch_hashes(id) ON DELETE CASCADE,
+    right_namespace_id INTEGER NOT NULL REFERENCES branch_hashes(id) ON DELETE CASCADE,
+
+    -- Since different codebases can have different variable names and such we also need to sandbox diffs by codebase owner
+    left_codebase_owner_user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    right_codebase_owner_user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+
+    diff JSONB NOT NULL,
+
+    PRIMARY KEY (left_namespace_id, right_namespace_id, left_codebase_owner_user_id, right_codebase_owner_user_id)
+);
+
+
+-- New table for coordinating background job for pre-computing diffs
+
+-- Table of all contributions which have been updated and may need their diffs re-computed
+CREATE TABLE contribution_diff_queue (
+  contribution_id UUID PRIMARY KEY REFERENCES contributions(id) ON DELETE CASCADE,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);

src/Share/BackgroundJobs.hs

@@ -1,10 +1,12 @@
 module Share.BackgroundJobs (startWorkers) where
 
 import Ki.Unlifted qualified as Ki
+import Share.BackgroundJobs.Diffs.ContributionDiffs qualified as ContributionDiffs
 import Share.BackgroundJobs.Monad (Background)
 import Share.BackgroundJobs.Search.DefinitionSync qualified as DefnSearch
 
 -- | Kicks off all background workers.
 startWorkers :: Ki.Scope -> Background ()
 startWorkers scope = do
   DefnSearch.worker scope
+  ContributionDiffs.worker scope

src/Share/BackgroundJobs/Diffs/ContributionDiffs.hs

@@ -0,0 +1,71 @@
+module Share.BackgroundJobs.Diffs.ContributionDiffs (worker) where
+
+import Control.Lens
+import Control.Monad.Except (ExceptT (..), runExceptT)
+import Ki.Unlifted qualified as Ki
+import Share.BackgroundJobs.Diffs.Queries qualified as DQ
+import Share.BackgroundJobs.Errors (reportError)
+import Share.BackgroundJobs.Monad (Background)
+import Share.BackgroundJobs.Workers (newWorker)
+import Share.Branch (Branch (..))
+import Share.Codebase qualified as Codebase
+import Share.Contribution (Contribution (..))
+import Share.IDs
+import Share.Metrics qualified as Metrics
+import Share.NamespaceDiffs (NamespaceDiffError (MissingEntityError))
+import Share.Postgres qualified as PG
+import Share.Postgres.Contributions.Queries qualified as ContributionsQ
+import Share.Postgres.Queries qualified as Q
+import Share.Prelude
+import Share.Utils.Logging qualified as Logging
+import Share.Web.Authorization qualified as AuthZ
+import Share.Web.Errors (EntityMissing (..), ErrorID (..))
+import Share.Web.Share.Diffs.Impl qualified as Diffs
+import Unison.Debug qualified as Debug
+import UnliftIO.Concurrent qualified as UnliftIO
+
+pollingIntervalSeconds :: Int
+pollingIntervalSeconds = 10
+
+worker :: Ki.Scope -> Background ()
+worker scope = do
+  authZReceipt <- AuthZ.backgroundJobAuthZ
+  newWorker scope "diffs:contributions" $ forever do
+    processDiffs authZReceipt >>= \case
+      Left e -> reportError e
+      Right _ -> pure ()
+    liftIO $ UnliftIO.threadDelay $ pollingIntervalSeconds * 1000000
+
+processDiffs :: AuthZ.AuthZReceipt -> Background (Either NamespaceDiffError ())
+processDiffs authZReceipt = Metrics.recordContributionDiffDuration . runExceptT $ do
+  Debug.debugLogM Debug.Temp "Background: Getting contributions to be diffed"
+  mayContributionId <- PG.runTransactionMode PG.ReadCommitted PG.ReadWrite $ do
+    DQ.claimContributionToDiff
+  Debug.debugM Debug.Temp "Background: contribution to be diffed: " mayContributionId
+  for_ mayContributionId (diffContribution authZReceipt)
+  case mayContributionId of
+    Just contributionId -> do
+      Logging.textLog ("Recomputed contribution diff: " <> tShow contributionId)
+        & Logging.withTag ("contribution-id", tShow contributionId)
+        & Logging.withSeverity Logging.Info
+        & Logging.logMsg
+      -- Keep processing releases until we run out of them.
+      either throwError pure =<< lift (processDiffs authZReceipt)
+    Nothing -> pure ()
+
+diffContribution :: AuthZ.AuthZReceipt -> ContributionId -> ExceptT NamespaceDiffError Background ()
+diffContribution authZReceipt contributionId = do
+  ( project,
+    newBranch@Branch {causal = newBranchCausalId},
+    oldBranch@Branch {causal = oldBranchCausalId}
+    ) <- ExceptT $ PG.tryRunTransaction $ do
+    Contribution {sourceBranchId = newBranchId, targetBranchId = oldBranchId, projectId} <- ContributionsQ.contributionById contributionId `whenNothingM` throwError (MissingEntityError $ EntityMissing (ErrorID "contribution:missing") "Contribution not found")
+    project <- Q.projectById projectId `whenNothingM` throwError (MissingEntityError $ EntityMissing (ErrorID "project:missing") "Project not found")
+    newBranch <- Q.branchById newBranchId `whenNothingM` throwError (MissingEntityError $ EntityMissing (ErrorID "branch:missing") "Source branch not found")
+    oldBranch <- Q.branchById oldBranchId `whenNothingM` throwError (MissingEntityError $ EntityMissing (ErrorID "branch:missing") "Target branch not found")
+    pure (project, newBranch, oldBranch)
+  let oldCodebase = Codebase.codebaseForProjectBranch authZReceipt project oldBranch
+  let newCodebase = Codebase.codebaseForProjectBranch authZReceipt project newBranch
+  -- This method saves the diff so it'll be there when we need it, so we don't need to do anything with it.
+  _ <- Diffs.diffCausals authZReceipt (oldCodebase, oldBranchCausalId) (newCodebase, newBranchCausalId)
+  pure ()

src/Share/BackgroundJobs/Diffs/Queries.hs

@@ -0,0 +1,43 @@
+module Share.BackgroundJobs.Diffs.Queries
+  ( submitContributionsToBeDiffed,
+    claimContributionToDiff,
+  )
+where
+
+import Data.Foldable (toList)
+import Data.Set (Set)
+import Share.IDs
+import Share.Postgres
+import Unison.Debug qualified as Debug
+
+submitContributionsToBeDiffed :: (QueryM m) => Set ContributionId -> m ()
+submitContributionsToBeDiffed contributions = do
+  Debug.debugM Debug.Temp "Submitting contributions to be diffed: " contributions
+  execute_
+    [sql|
+    WITH new_contributions(contribution_id) AS (
+      SELECT * FROM ^{singleColumnTable (toList contributions)}
+    )
+    INSERT INTO contribution_diff_queue (contribution_id)
+      SELECT nc.contribution_id FROM new_contributions nc
+      ON CONFLICT DO NOTHING
+    |]
+
+-- | Claim the oldest contribution in the queue to be diffed.
+claimContributionToDiff :: Transaction e (Maybe ContributionId)
+claimContributionToDiff = do
+  query1Col
+    [sql|
+    WITH chosen_contribution(contribution_id) AS (
+      SELECT q.contribution_id
+      FROM contribution_diff_queue q
+      ORDER BY q.created_at ASC
+      LIMIT 1
+      -- Skip any that are being synced by other workers.
+      FOR UPDATE SKIP LOCKED
+    )
+    DELETE FROM contribution_diff_queue
+      USING chosen_contribution
+      WHERE contribution_diff_queue.contribution_id = chosen_contribution.contribution_id
+    RETURNING chosen_contribution.contribution_id
+    |]

src/Share/Metrics.hs

@@ -10,6 +10,7 @@ module Share.Metrics
     tickUserSignup,
     recordBackgroundImportDuration,
     recordDefinitionSearchIndexDuration,
+    recordContributionDiffDuration,
   )
 where
 
@@ -398,6 +399,18 @@ definitionSearchIndexDurationSeconds =
         "definition_search_indexing_duration_seconds"
         "The time it took to index a release for definition search"
 
+{-# NOINLINE contributionDiffDurationSeconds #-}
+contributionDiffDurationSeconds :: Prom.Vector Prom.Label2 Prom.Histogram
+contributionDiffDurationSeconds =
+  Prom.unsafeRegister $
+    Prom.vector ("deployment", "service") $
+      Prom.histogram info Prom.defaultBuckets
+  where
+    info =
+      Prom.Info
+        "contribution_diff_duration_seconds"
+        "The time it took to compute a contribution diff"
+
 timeActionIntoHistogram :: (Prom.Label l, MonadUnliftIO m) => (Prom.Vector l Prom.Histogram) -> l -> m c -> m c
 timeActionIntoHistogram histogram l m = do
   UnliftIO.bracket start end \_ -> m
@@ -416,3 +429,6 @@ recordBackgroundImportDuration = timeActionIntoHistogram backgroundImportDuratio
 -- | Record the duration of a background import.
 recordDefinitionSearchIndexDuration :: (MonadUnliftIO m) => m r -> m r
 recordDefinitionSearchIndexDuration = timeActionIntoHistogram definitionSearchIndexDurationSeconds (deployment, service)
+
+recordContributionDiffDuration :: (MonadUnliftIO m) => m r -> m r
+recordContributionDiffDuration = timeActionIntoHistogram contributionDiffDurationSeconds (deployment, service)