Skip to content

Commit eeeda7e

Browse files
ChrisPennerChrisPenner
committed
Bump
1 parent c0f1283 commit eeeda7e

File tree

3 files changed

+49
-39
lines changed

3 files changed

+49
-39
lines changed

‎Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
SHARE_PROJECT_ROOT := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
44
export SHARE_PROJECT_ROOT
55
UNAME := $(shell uname)
6-
STACK_FLAGS := "--fast"
6+
STACK_FLAGS := --fast
77
dist_dir := $(shell stack path | awk '/^dist-dir/{print $$2}')
88
exe_name := share-api
99
exe := $(dist_dir)/build/$(exe_name)/$(exe_name)

‎src/Share/Web/UCM/SyncV3/Impl.hs‎

Lines changed: 44 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@ import U.Codebase.Sqlite.Orphans ()
2929
import Unison.Debug qualified as Debug
3030
import Unison.Hash32 (Hash32)
3131
import Unison.Share.API.Hash (HashJWT, HashJWTClaims (..))
32+
import Unison.Share.API.Hash qualified as HashJWT
3233
import Unison.SyncV3.Types
34+
import Unison.SyncV3.Utils (entityDependencies)
3335
import Unison.Util.Websockets (Queues (..), withQueues)
3436
import UnliftIO qualified
3537
import UnliftIO.STM
@@ -90,6 +92,7 @@ doSyncEmitter mayCallerUserId conn = do
9092
syncState <- case initMsg of
9193
ReceiverInitStream initMsg -> initialize onErr mayCallerUserId initMsg
9294
other -> onErr $ InitializationError ("Expected ReceiverInitStream message, got: " <> tShow other)
95+
Debug.debugLogM Debug.Temp "Initialized sync state, starting sync process."
9396
lift (shareEmitter syncState q)
9497
>>= maybe (pure ()) (onErr)
9598
where
@@ -102,24 +105,29 @@ doSyncEmitter mayCallerUserId conn = do
102105
Right <$> action (fmap absurd . cc . Left)
103106
-- If we get an error, send it to the client then shut down.
104107
handleErr ::
108+
(Show err) =>
105109
Queues (MsgOrError err a) o ->
106110
WebApp (Either err ()) ->
107111
WebApp ()
108112
handleErr (Queues {send, shutdown}) action = do
109113
action >>= \case
110114
Left err -> do
115+
Debug.debugM Debug.Temp "Sync error, shutting down: " err
111116
atomically $ do
112117
send (Err err)
113118
liftIO $ shutdown
114119
Right r -> pure r
115120

116121
initialize :: (forall x. SyncError -> SyncM x) -> (Maybe UserId) -> InitMsg HashJWT -> SyncM (SyncState sh Hash32)
117122
initialize onErr caller InitMsg {initMsgRootCausal, initMsgBranchRef} = do
123+
let decoded = HashJWT.decodeHashJWT initMsgRootCausal
124+
Debug.debugM Debug.Temp "Decoded root causal hash jwt" decoded
125+
Debug.debugM Debug.Temp "Caller: " caller
118126
HashJWTClaims {hash = initialCausalHash} <-
119127
lift (HashJWT.verifyHashJWT caller initMsgRootCausal) >>= \case
120128
Right ch -> pure ch
121129
Left err -> onErr $ HashJWTVerificationError (AuthN.authErrMsg err)
122-
validRequestsVar <- newTVarIO Set.empty
130+
validRequestsVar <- newTVarIO (Set.singleton (CausalEntity, initialCausalHash))
123131
requestedEntitiesVar <- newTVarIO (Set.singleton (CausalEntity, initialCausalHash))
124132
entitiesAlreadySentVar <- newTVarIO Set.empty
125133
(lift . runExceptT $ codebaseForBranchRef initMsgBranchRef) >>= \case
@@ -160,47 +168,45 @@ shareEmitter SyncState {requestedEntitiesVar, entitiesAlreadySentVar, validReque
160168
let onErrSTM :: SyncError -> STM ()
161169
onErrSTM e = do
162170
UnliftIO.putTMVar errVar e
171+
172+
Debug.debugLogM Debug.Temp "Launching workers"
163173
Ki.fork scope $ sendWorker onErrSTM
164174
Ki.fork scope $ receiveWorker onErrSTM
175+
Debug.debugLogM Debug.Temp "Waiting on errors or completion..."
165176
atomically ((Ki.awaitAll scope $> Nothing) <|> (Just <$> UnliftIO.takeTMVar errVar))
166177
where
167178
sendWorker :: (SyncError -> STM ()) -> WebApp ()
168179
sendWorker onErrSTM = forever $ do
169-
reqs <- atomically $ do
180+
(validRequests, reqs) <- atomically $ do
170181
reqs <- readTVar requestedEntitiesVar
182+
writeTVar requestedEntitiesVar Set.empty
183+
alreadySent <- readTVar entitiesAlreadySentVar
184+
Debug.debugM Debug.Temp "Processing Requested entities: " reqs
171185
validRequests <- readTVar validRequestsVar
172-
let forbiddenRequests = Set.difference reqs validRequests
173-
validRequests <-
174-
if not (Set.null forbiddenRequests)
175-
then do
176-
onErrSTM (ForbiddenEntityRequest forbiddenRequests)
177-
pure $ Set.difference validRequests forbiddenRequests
178-
else do
179-
pure validRequests
180-
guard (not $ Set.null validRequests)
181-
-- TODO: Add reasonable batch sizes
182-
modifyTVar' requestedEntitiesVar (const Set.empty)
183-
sent <- readTVar entitiesAlreadySentVar
184-
let unsent = Set.difference reqs sent
185-
guard (not $ Set.null unsent)
186-
pure unsent
187-
newEntities <- fetchEntities codebase reqs
188-
-- let hashMappings :: Map HashTag Hash32
189-
-- hashMappings =
190-
-- newEntities
191-
-- & toListOf (folded . entityHashesGetter_)
192-
-- & Map.fromList
193-
-- atomically $ do
194-
-- alreadyMapped <- readTVar mappedHashesVar
195-
-- let newMappings = Map.difference hashMappings alreadyMapped
196-
-- modifyTVar' mappedHashesVar (Map.union newMappings)
197-
-- send (HashMappingsMsg (HashMappings (newMappings)))
198-
199-
atomically $ do
200-
let newHashes = setOf (folded . to (entityKind &&& entityHash)) newEntities
201-
modifyTVar' entitiesAlreadySentVar (Set.union newHashes)
202-
for newEntities \entity -> do
203-
send $ Msg (EmitterEntityMsg entity)
186+
pure (validRequests, reqs `Set.difference` alreadySent)
187+
let forbiddenRequests = Set.difference reqs validRequests
188+
validatedRequests <-
189+
if not (Set.null forbiddenRequests)
190+
then do
191+
atomically (onErrSTM (ForbiddenEntityRequest forbiddenRequests))
192+
pure $ Set.difference validRequests forbiddenRequests
193+
else do
194+
pure validRequests
195+
Debug.debugM Debug.Temp "Validated requests: " validatedRequests
196+
when (not $ Set.null validatedRequests) $ do
197+
Debug.debugM Debug.Temp "Fetching Entities." validatedRequests
198+
newEntities <- fetchEntities codebase validatedRequests
199+
Debug.debugM Debug.Temp "Fetched entities: " (length newEntities)
200+
-- Do work outside of transactions to avoid conflicts
201+
deps <- UnliftIO.evaluate $ foldMap entityDependencies newEntities
202+
Debug.debugLogM Debug.Temp "Adding new valid requests"
203+
atomically $ modifyTVar' validRequestsVar (\s -> Set.union s deps)
204+
Debug.debugM Debug.Temp "Sending entities: " (length newEntities)
205+
atomically $ do
206+
let newHashes = setOf (folded . to (entityKind &&& entityHash)) newEntities
207+
modifyTVar' entitiesAlreadySentVar (Set.union newHashes)
208+
for_ newEntities \entity -> do
209+
send $ Msg (EmitterEntityMsg entity)
204210

205211
receiveWorker :: (SyncError -> STM ()) -> WebApp ()
206212
receiveWorker onErrSTM = forever $ do
@@ -209,8 +215,12 @@ shareEmitter SyncState {requestedEntitiesVar, entitiesAlreadySentVar, validReque
209215
Err err -> onErrSTM err
210216
Msg (ReceiverInitStream {}) -> onErrSTM (InitializationError "Received duplicate ReceiverInitStream message")
211217
Msg (ReceiverEntityRequest (EntityRequestMsg {hashes})) -> do
218+
Debug.debugM Debug.Temp "Got new entity requests" hashes
212219
modifyTVar' requestedEntitiesVar (\s -> Set.union s (Set.fromList hashes))
213220

214221
fetchEntities :: CodebaseEnv -> Set (EntityKind, Hash32) -> WebApp (Vector (Entity Hash32 Text))
215222
fetchEntities codebase reqs = do
216223
PG.runTransaction $ Q.fetchSerialisedEntities codebase reqs
224+
225+
-- entityDependencies :: Entity hash text -> Set (EntityKind, Hash32)
226+
-- entityDependencies (Entity {entityData}) = do

‎src/Share/Web/UCM/SyncV3/Queries.hs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,13 +23,13 @@ fetchSerialisedEntities (CodebaseEnv {codebaseOwner}) requestedEntities =
2323
(SELECT req.kind, bytes.bytes, ch.base32, cd.depth
2424
FROM requested req
2525
JOIN component_hashes ch ON req.hash = ch.base32
26-
JOIN serialized_components sc ON sc.user_id = #{codebaseOwner} AND ch.component_hash_id = sc.component_hash_id
26+
JOIN serialized_components sc ON sc.user_id = #{codebaseOwner} AND ch.id = sc.component_hash_id
2727
JOIN bytes ON sc.bytes_id = bytes.id
2828
JOIN component_depth cd ON ch.id = cd.component_hash_id
2929
WHERE req.kind = 'component'
3030
)
3131
UNION ALL
32-
(SELECT req.kind, bytes.bytes, ap.patch_hash, pd.depth
32+
(SELECT req.kind, bytes.bytes, req.hash, pd.depth
3333
FROM requested req
3434
JOIN patches p ON req.hash = p.hash
3535
JOIN serialized_patches sp ON p.id = sp.patch_id
@@ -38,7 +38,7 @@ fetchSerialisedEntities (CodebaseEnv {codebaseOwner}) requestedEntities =
3838
WHERE req.kind = 'patch'
3939
)
4040
UNION ALL
41-
(SELECT req.kind, bytes.bytes, an.namespace_hash, nd.depth
41+
(SELECT req.kind, bytes.bytes, req.hash, nd.depth
4242
FROM requested req
4343
JOIN branch_hashes bh ON req.hash = bh.base32
4444
JOIN serialized_namespaces sn ON bh.id = sn.namespace_hash_id
@@ -49,7 +49,7 @@ fetchSerialisedEntities (CodebaseEnv {codebaseOwner}) requestedEntities =
4949
UNION ALL
5050
-- TODO: Should probably join in a batch of causal spines here too
5151
-- to improve parallelism and avoid long-spine bottlenecks.
52-
(SELECT req.kind, bytes.bytes, tc.causal_hash, cd.depth
52+
(SELECT req.kind, bytes.bytes, req.hash, cd.depth
5353
FROM requested req
5454
JOIN causals c ON req.hash = c.hash
5555
JOIN serialized_causals sc ON c.id = sc.causal_id

0 commit comments

Comments
 (0)