diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,16 @@ ## Unreleased +## 0.1.0.2 - 2026-03-23 + +### Fixed + +- Fixed a bug where an async exception delivered after `loadIntoCache` returned + (but before `signalCacheLoaded` executed) would leave waiting threads permanently + blocked on the semaphore MVar. The `signalCacheLoaded` call is now inside + `uninterruptibleMask_` alongside `adjustCacheEntryOnLoadError`, ensuring the + semaphore is always signalled regardless of async exceptions. + ## 0.1.0.1 - 2026-03-22 ### Fixed diff --git a/shared-resource-cache.cabal b/shared-resource-cache.cabal --- a/shared-resource-cache.cabal +++ b/shared-resource-cache.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: shared-resource-cache -version: 0.1.0.1 +version: 0.1.0.2 synopsis: A thread-safe cache for sharing resources across threads with automatic expiry description: A cache designed for guaranteeing that different threads can share the same reference to a resource. For example, it allows threads to communicate via a shared <>. diff --git a/src/Data/SharedResourceCache.hs b/src/Data/SharedResourceCache.hs --- a/src/Data/SharedResourceCache.hs +++ b/src/Data/SharedResourceCache.hs @@ -71,6 +71,8 @@ -- -- If the item is already loaded into the cache, it is returned. If it is not yet loaded, the first thread requesting this resource to -- acquire a MVar will take ownership of loading it and signal to any other threads waiting for the resource that it has been loaded. +-- If the IO action throws an error or results a 'Left', the MVar is removed from the cache and another waiting thread (or a future call if there are +-- no current calls) can pick up the MVar to load the resource -- -- This action is executed in the 'MonadResource' monad typeclass -- context so that when the resource is freed, the cache can mark the cache item as having one less sharer. @@ -83,6 +85,7 @@ (releaseKey, cacheEntry) <- allocate (allocateResource resourceCache resourceId) (deAllocateResource resourceCache) return (releaseKey, cacheItem <$> cacheEntry) where + allocateResource :: SharedResourceCache err a -> Text -> IO (Either err (CacheItem a)) allocateResource = loadCacheableResource diff --git a/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs b/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs --- a/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs +++ b/src/Data/SharedResourceCache/Internal/ExpiringSharedResourceCache.hs @@ -65,24 +65,26 @@ Nothing -> loadFreshlyIntoCache resourceCache resourceId loadFreshlyIntoCache :: SharedResourceCache err a -> Text -> IO (Either err (CacheItem a)) - loadFreshlyIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = - -- We run this operation in mask so that we aren't left with a permanent loading claim in the cache if the thread is interrupted - mask $ (\restore -> do - maybeSemaphore <- takeOwnershipOfLoad resourceCache resourceId + loadFreshlyIntoCache resourceCache@(SharedResourceCache cache _ loadResourceOp _ _ _) resourceId = do + maybeSemaphore <- takeOwnershipOfLoad resourceCache resourceId - case maybeSemaphore of - -- Already loaded or loading in another thread, recursively start again - Nothing -> restore $ loadCacheableResource resourceCache resourceId + case maybeSemaphore of + -- Already loaded or loading in another thread, recursively start again + Nothing -> loadCacheableResource resourceCache resourceId - -- We've claimed ownership of loading the item into the cache - Just semaphore -> do - result <- restore (loadIntoCache resourceCache resourceId semaphore) - `onException` uninterruptibleMask_ (adjustCacheEntryOnLoadError resourceCache resourceId >> signalCacheLoaded semaphore) + -- We've claimed ownership of loading the item into the cache + Just semaphore -> do + -- We use uninterruptibleMask so that an async exception can't stop the sempahore remaining in the cache blocking any readers of the resource + -- from getting the resource. We're in the context of resourcet's 'allocate' so we're otherwise 'masked' apart from on blocking operations + result <- loadIntoCache resourceCache resourceId semaphore + `onException` uninterruptibleMask_ (adjustCacheEntryOnLoadError resourceCache resourceId >> signalCacheLoaded semaphore) - when (isLeft result) (uninterruptibleMask_ (adjustCacheEntryOnLoadError resourceCache resourceId)) + uninterruptibleMask_ $ do + when (isLeft result) (adjustCacheEntryOnLoadError resourceCache resourceId) signalCacheLoaded semaphore - pure result - ) + + pure result + where signalCacheLoaded semaphore = putMVar semaphore () @@ -118,14 +120,12 @@ adjustCacheEntryOnLoadError :: SharedResourceCache err a -> Text -> IO () adjustCacheEntryOnLoadError resourceCache@(SharedResourceCache cache _ _ _ _ _) resourceId = do - now <- getCurrentTime atomically $ do result <- M.lookup resourceId cache case result of - -- Error occurred before we were able to load the cache entry - delete it + -- Error occurred before we were able to load the cache entry - delete it so that another thread can claim + -- the semaphore to try loading it Just (LoadingEntry _) -> M.delete resourceId cache - -- Error occurred after we loaded the cache entry meaning although we won't subscribe other threads may - -- until the entry is expired - Just (LoadedEntry item) -> handleSharerLeaveSTM resourceCache item resourceId now + -- It shouldn't be possible for a loaded entry to enter the cache and _then_ an exception to occur _ -> pure ()