bolty 0.1.0.1 → 0.1.0.2

raw patch · 8 files changed

+243/−17 lines, 8 filesdep −packstreamdep ~packstream-boltPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependencies removed: packstream

Dependency ranges changed: packstream-bolt

API changes (from Hackage documentation)

Files

• bolty.cabal +3/−3
• changelog.md +14/−0
• integration/Main.hs +108/−0
• src/Database/Bolty.hs +6/−0
• src/Database/Bolty/Pool.hs +74/−13
• src/Database/Bolty/Routing.hs +33/−0
• src/Database/Bolty/Session.hs +2/−1
• test/Main.hs +3/−0

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           bolty-version:        0.1.0.1+version:        0.1.0.2 synopsis:       Haskell driver for Neo4j (BOLT protocol 4.4-5.4) description:    Native Haskell driver for Neo4j graph database using the BOLT protocol.                 Supports BOLT versions 4.4 through 5.4 with connection pooling,@@ -150,7 +150,7 @@     , extra >=1.7 && <1.9     , mtl >=2.2 && <2.4     , network >=3.1 && <3.3-    , packstream+    , packstream-bolt     , persist ==1.0.*     , sandwich     , split ==0.2.*@@ -202,7 +202,7 @@     , extra >=1.7 && <1.9     , mtl >=2.2 && <2.4     , network >=3.1 && <3.3-    , packstream+    , packstream-bolt     , persist ==1.0.*     , sandwich     , split ==0.2.*

@@ -6,6 +6,20 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.1.0.2++- Fix pool crash on machines with more than 10 CPU cores (set numStripes to 1)+- Add `acquireConnection`, `releaseConnection`, `releaseConnectionOnError` for manual connection lifetime management+- Add `acquireRoutingConnection` for routing-aware manual connection acquisition+- Expose `Session(..)` and `SessionPool(..)` constructors+- Fix `sigLocalDateTime` tag (was 0x66, now 0x64)+- Send `patch_bolt: ["utc"]` in HELLO for BOLT 4.x connections+- Complete `psToBolt`/`boltToPs` for all 13 structure types++## 0.1.0.1++- Fix dependency name: packstream renamed to packstream-bolt+ ## 0.1.0.0  - Initial release

@@ -1560,6 +1560,112 @@   -- ================================================================+-- Pool configuration tests+-- ================================================================++poolConfigTestSuite :: TopSpec+poolConfigTestSuite = do+  let getConf = mkGetConfig neo4j5Config++  describe "Pool configuration" $ do++    it "pool works with maxConnections = 1" $ do+      cfg <- liftIO getConf+      let pc = defaultPoolConfig { maxConnections = 1 }+      pool <- liftIO $ createPool cfg pc+      result <- liftIO $ withConnection pool $ \p ->+        queryIO p "RETURN 1 AS n"+      case asInt (V.head $ V.head result) of+        Just n  -> n `shouldBe` 1+        Nothing -> expectationFailure "Expected 1"+      liftIO $ destroyPool pool++    it "pool works with maxConnections = 2" $ do+      cfg <- liftIO getConf+      let pc = defaultPoolConfig { maxConnections = 2 }+      pool <- liftIO $ createPool cfg pc+      result <- liftIO $ withConnection pool $ \p ->+        queryIO p "RETURN 42 AS n"+      case asInt (V.head $ V.head result) of+        Just n  -> n `shouldBe` 42+        Nothing -> expectationFailure "Expected 42"+      liftIO $ destroyPool pool++    it "pool with maxConnections = 1 serializes access" $ do+      cfg <- liftIO getConf+      let pc = defaultPoolConfig { maxConnections = 1 }+      pool <- liftIO $ createPool cfg pc+      r1 <- liftIO $ withConnection pool $ \p -> queryIO p "RETURN 1 AS n"+      r2 <- liftIO $ withConnection pool $ \p -> queryIO p "RETURN 2 AS n"+      case (asInt (V.head $ V.head r1), asInt (V.head $ V.head r2)) of+        (Just 1, Just 2) -> pure ()+        _                -> expectationFailure "Expected 1 and 2"+      liftIO $ destroyPool pool++    it "withTransaction' works with maxConnections = 1" $ do+      cfg <- liftIO getConf+      let pc = defaultPoolConfig { maxConnections = 1 }+      pool <- liftIO $ createPool cfg pc+      liftIO $ withTransaction' pool $ \conn -> do+        _ <- queryIO conn "RETURN 1"+        pure ()+      liftIO $ destroyPool pool++    it "acquireConnection and releaseConnection work" $ do+      cfg <- liftIO getConf+      pool <- liftIO $ createPool cfg defaultPoolConfig+      coc <- liftIO $ acquireConnection pool+      result <- liftIO $ queryIO (cocConnection coc) "RETURN 99 AS n"+      liftIO $ releaseConnection coc+      case asInt (V.head $ V.head result) of+        Just n  -> n `shouldBe` 99+        Nothing -> expectationFailure "Expected 99"+      liftIO $ destroyPool pool++    it "releaseConnectionOnError discards bad connection" $ do+      cfg <- liftIO getConf+      pool <- liftIO $ createPool cfg defaultPoolConfig+      coc <- liftIO $ acquireConnection pool+      -- Simulate an error scenario: release as error+      liftIO $ releaseConnectionOnError coc+      -- Pool should still work — next checkout gets a fresh connection+      result <- liftIO $ withConnection pool $ \p ->+        queryIO p "RETURN 1 AS n"+      case asInt (V.head $ V.head result) of+        Just n  -> n `shouldBe` 1+        Nothing -> expectationFailure "Expected 1"+      liftIO $ destroyPool pool++    it "multiple acquireConnection calls work" $ do+      cfg <- liftIO getConf+      let pc = defaultPoolConfig { maxConnections = 3 }+      pool <- liftIO $ createPool cfg pc+      coc1 <- liftIO $ acquireConnection pool+      coc2 <- liftIO $ acquireConnection pool+      r1 <- liftIO $ queryIO (cocConnection coc1) "RETURN 1 AS n"+      r2 <- liftIO $ queryIO (cocConnection coc2) "RETURN 2 AS n"+      liftIO $ releaseConnection coc1+      liftIO $ releaseConnection coc2+      case (asInt (V.head $ V.head r1), asInt (V.head $ V.head r2)) of+        (Just 1, Just 2) -> pure ()+        _                -> expectationFailure "Expected 1 and 2"+      liftIO $ destroyPool pool++    it "poolCounters tracks acquisitions" $ do+      cfg <- liftIO getConf+      pool <- liftIO $ createPool cfg defaultPoolConfig+      pc0 <- liftIO $ poolCounters pool+      pcTotalAcqs pc0 `shouldBe` 0+      _ <- liftIO $ withConnection pool $ \p -> queryIO p "RETURN 1"+      pc1 <- liftIO $ poolCounters pool+      pcTotalAcqs pc1 `shouldBe` 1+      _ <- liftIO $ withConnection pool $ \p -> queryIO p "RETURN 2"+      pc2 <- liftIO $ poolCounters pool+      pcTotalAcqs pc2 `shouldBe` 2+      liftIO $ destroyPool pool+++-- ================================================================ -- Main -- ================================================================ @@ -1602,6 +1708,8 @@   serverHintTestSuite   -- Decode tests (Neo4j 5)   decodeTestSuite+  -- Pool configuration tests+  poolConfigTestSuite   -- Temporal type tests   temporalTestSuite "Neo4j 5" neo4j5Config   temporalTestSuite "Neo4j 4.4" neo4j44Config

@@ -69,6 +69,10 @@   , createPool   , destroyPool   , withConnection+  , CheckedOutConnection(..)+  , acquireConnection+  , releaseConnection+  , releaseConnectionOnError   , poolCounters     -- * Routing   , AccessMode(..)@@ -79,6 +83,7 @@   , createRoutingPool   , destroyRoutingPool   , withRoutingConnection+  , acquireRoutingConnection   , withRoutingTransaction   , invalidateRoutingTable   , getRoutingTable@@ -166,6 +171,7 @@ import           Database.Bolty.Routing (AccessMode(..), RoutingPool, RoutingPoolConfig(..),                                         defaultRoutingPoolConfig, createRoutingPool,                                         destroyRoutingPool, withRoutingConnection,+                                        acquireRoutingConnection,                                         withRoutingTransaction, invalidateRoutingTable,                                         getRoutingTable) import           Database.Bolty.Session (Session, SessionConfig(..), defaultSessionConfig,

@@ -10,6 +10,10 @@   , createPool   , destroyPool   , withConnection+  , CheckedOutConnection(..)+  , acquireConnection+  , releaseConnection+  , releaseConnectionOnError   , poolCounters   ) where @@ -126,7 +130,8 @@           min idleTimeout (fromIntegral (serverSecs - 5))         _ -> idleTimeout   P.close probe-  let poolCfg = Pool.defaultPoolConfig+  let poolCfg = Pool.setNumStripes (Just 1)+              $ Pool.defaultPoolConfig                   (P.connect cfg)           -- create                   P.close                   -- destroy                   effectiveIdle             -- idle timeout (seconds)@@ -187,27 +192,83 @@           decActive           restore $ go (n - 1) canRetryDead -    -- | Check if a connection needs a ping based on the validation strategy.-    needsPing :: ValidationStrategy -> Connection -> IO Bool-    needsPing AlwaysPing _    = pure True-    needsPing NeverPing _     = pure False-    needsPing (PingIfIdle secs) conn = do-      now <- getMonotonicTimeNSec-      lastAct <- connectionLastActivity conn-      let idleNs = now - lastAct-      let thresholdNs = fromIntegral secs * 1_000_000_000-      pure (idleNs >= thresholdNs)-     throwIOSome :: SomeException -> IO a     throwIOSome = throwIO' where       throwIO' :: SomeException -> IO a       throwIO' = Control.Exception.throwIO -    -- | Check if an exception is a connection-level error (not a query error).     isConnectionError :: SomeException -> Bool     isConnectionError e = case fromException e :: Maybe Error of       Just (NonboltyError _) -> True       _                      -> False+++-- | Check if a connection needs a ping based on the validation strategy.+needsPing :: ValidationStrategy -> Connection -> IO Bool+needsPing AlwaysPing _    = pure True+needsPing NeverPing _     = pure False+needsPing (PingIfIdle secs) conn = do+  now <- getMonotonicTimeNSec+  lastAct <- connectionLastActivity conn+  let idleNs = now - lastAct+  let thresholdNs = fromIntegral secs * 1_000_000_000+  pure (idleNs >= thresholdNs)+++-- | A checked-out connection handle, bundling the connection with its+-- local pool reference for proper release.+type CheckedOutConnection :: Type+data CheckedOutConnection = CheckedOutConnection+  { cocConnection :: !Connection+  , cocLocalPool  :: !(Pool.LocalPool Connection)+  , cocBoltPool   :: !BoltPool+  }+++-- | Acquire a validated connection from the pool. The caller is responsible+-- for releasing it with 'releaseConnection' or 'releaseConnectionOnError'.+--+-- This is the low-level primitive behind 'withConnection'. Prefer+-- 'withConnection' for simple request-response patterns.+-- Use 'acquireConnection' when you need the connection to outlive a callback+-- (e.g. for streaming with @bracketIO@).+acquireConnection :: BoltPool -> IO CheckedOutConnection+acquireConnection bp@BoltPool{bpPool, bpMaxRetries, bpValidation, bpActiveConns, bpTotalAcqs, bpTotalWaitNs} =+    go (bpMaxRetries + 1)+  where+    go 0 = fail "acquireConnection: no healthy connection available"+    go n = do+      t0 <- getMonotonicTimeNSec+      (conn, localPool) <- Pool.takeResource bpPool+      t1 <- getMonotonicTimeNSec+      atomicModifyIORef' bpActiveConns $ \x -> (x + 1, ())+      atomicModifyIORef' bpTotalAcqs $ \x -> (x + 1, ())+      atomicModifyIORef' bpTotalWaitNs $ \x -> (x + (t1 - t0), ())+      shouldPing <- needsPing bpValidation conn+      healthy <- if shouldPing then P.ping conn else pure True+      if healthy+        then pure CheckedOutConnection+              { cocConnection = conn, cocLocalPool = localPool, cocBoltPool = bp }+        else do+          Pool.destroyResource bpPool localPool conn+          atomicModifyIORef' bpActiveConns $ \x -> (x - 1, ())+          go (n - 1)+++-- | Release a connection back to the pool after successful use.+releaseConnection :: CheckedOutConnection -> IO ()+releaseConnection CheckedOutConnection{cocConnection, cocLocalPool, cocBoltPool} = do+  touchConnection cocConnection+  Pool.putResource cocLocalPool cocConnection+  atomicModifyIORef' (bpActiveConns cocBoltPool) $ \x -> (x - 1, ())+++-- | Release a connection after an error, destroying it instead of returning+-- it to the pool (since it may be in a bad state).+releaseConnectionOnError :: CheckedOutConnection -> IO ()+releaseConnectionOnError CheckedOutConnection{cocConnection, cocLocalPool, cocBoltPool} = do+  Pool.destroyResource (bpPool cocBoltPool) cocLocalPool cocConnection+  atomicModifyIORef' (bpActiveConns cocBoltPool) $ \x -> (x - 1, ())   -- | Read a snapshot of the pool's lifetime counters.

@@ -12,6 +12,7 @@   , createRoutingPool   , destroyRoutingPool   , withRoutingConnection+  , acquireRoutingConnection   , withRoutingTransaction   , invalidateRoutingTable     -- * Internals (exported for testing)@@ -170,6 +171,38 @@   when (V.null addrs) $     throwIO $ RoutingTableError $ "No servers available for " <> T.pack (show mode)   tryAddresses rp addrs (V.length addrs) Nothing action+++-- | Acquire a routed connection by access mode. Returns a+-- 'CheckedOutConnection' that must be released by the caller.+-- Tries addresses in round-robin order, failing over on unavailable servers.+acquireRoutingConnection :: HasCallStack => RoutingPool -> AccessMode -> IO CheckedOutConnection+acquireRoutingConnection rp mode = do+  rt <- getOrRefreshTable rp+  let addrs = case mode of+        ReadAccess  -> readers rt+        WriteAccess -> writers rt+  when (V.null addrs) $+    throwIO $ RoutingTableError $ "No servers available for " <> T.pack (show mode)+  tryAcquireAddresses rp addrs (V.length addrs) Nothing+++-- | Try addresses in round-robin order for acquire, failing over on connection errors.+tryAcquireAddresses :: HasCallStack+                    => RoutingPool -> V.Vector Text -> Int -> Maybe SomeException -> IO CheckedOutConnection+tryAcquireAddresses _rp _addrs 0 (Just lastErr) = throwIO lastErr+tryAcquireAddresses _rp _addrs 0 Nothing =+  throwIO $ RoutingTableError "All servers unavailable"+tryAcquireAddresses rp addrs remaining _lastErr = do+  addr <- roundRobin rp addrs+  pool <- getOrCreatePool rp addr+  result <- try $ acquireConnection pool+  case result of+    Right coc -> pure coc+    Left (e :: SomeException)+      | remaining > 1, isServerUnavailable e ->+          tryAcquireAddresses rp addrs (remaining - 1) (Just e)+      | otherwise -> throwIO e   -- | Try addresses in round-robin order, failing over on connection errors.

@@ -6,7 +6,8 @@   , getBookmarks   , updateBookmark     -- * Session-  , Session+  , Session(..)+  , SessionPool(..)   , SessionConfig(..)   , defaultSessionConfig   , createSession

@@ -462,6 +462,9 @@   it "defaultPoolConfig has idleTimeout == 60" $ do     idleTimeout defaultPoolConfig `shouldBe` 60 +  it "defaultPoolConfig maxConnections >= 1 (required by resource-pool)" $ do+    (maxConnections defaultPoolConfig >= 1) `shouldBe` True+  -- = Retry config and isTransient tests