diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## [Unreleased] +## 0.23.0.0 — 2026-10-30 + +### Breaking Changes + +- Require `kiroku-store ^>=0.13.0.1` from + mori://shinzui/kiroku/packages/kiroku-store and align internal dependencies with + the shared release. The optional composition example uses + `kiroku-metrics ^>=0.4.0.1` from mori://shinzui/kiroku/packages/kiroku-metrics. + +### Bug Fixes + +- Close accepted sockets even when shutdown prevents their Warp connection + workers from starting, avoiding idle connections surviving server shutdown. + ## 0.22.0.0 — 2026-10-27 ### Breaking Changes diff --git a/keiro-ops-http.cabal b/keiro-ops-http.cabal --- a/keiro-ops-http.cabal +++ b/keiro-ops-http.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-ops-http -version: 0.22.0.0 +version: 0.23.0.0 synopsis: Bounded HTTP transport for the Keiro operational command tree @@ -78,10 +78,10 @@ hasql >=1.10 && <1.11, hasql-pool >=1.2 && <1.5, http-types >=0.12 && <0.13, - keiro ^>=0.22.0.0, - keiro-migrations ^>=0.22.0.0, - keiro-ops ^>=0.22.0.0, - kiroku-store ^>=0.12.0.0, + keiro ^>=0.23.0.0, + keiro-migrations ^>=0.23.0.0, + keiro-ops ^>=0.23.0.0, + kiroku-store ^>=0.13.0.1, network >=3.1 && <3.3, network-uri >=2.6.4.2 && <2.7, optparse-applicative >=0.19 && <0.20, @@ -107,10 +107,10 @@ async >=2.2 && <2.3, base >=4.21 && <5, hasql >=1.10 && <1.11, - keiro-migrations ^>=0.22.0.0, - keiro-ops ^>=0.22.0.0, + keiro-migrations ^>=0.23.0.0, + keiro-ops ^>=0.23.0.0, keiro-ops-http, - kiroku-store ^>=0.12.0.0, + kiroku-store ^>=0.13.0.1, optparse-applicative >=0.19 && <0.20, text >=2.1 && <2.2, @@ -133,10 +133,10 @@ build-depends: async >=2.2 && <2.3, base >=4.21 && <5, - keiro-ops ^>=0.22.0.0, + keiro-ops ^>=0.23.0.0, keiro-ops-http, - kiroku-metrics ^>=0.4.0.0, - kiroku-store ^>=0.12.0.0, + kiroku-metrics ^>=0.4.0.1, + kiroku-store ^>=0.13.0.1, optparse-applicative >=0.19 && <0.20, text >=2.1 && <2.2, @@ -165,11 +165,11 @@ http-client >=0.7 && <0.8, http-types >=0.12 && <0.13, jitsurei, - keiro ^>=0.22.0.0, - keiro-ops ^>=0.22.0.0, + keiro ^>=0.23.0.0, + keiro-ops ^>=0.23.0.0, keiro-ops-http, - keiro-test-support ^>=0.22.0.0, - kiroku-store ^>=0.12.0.0, + keiro-test-support ^>=0.23.0.0, + kiroku-store ^>=0.13.0.1, process >=1.6 && <1.7, text >=2.1 && <2.2, time >=1.12 && <1.15, @@ -205,14 +205,14 @@ http-client >=0.7 && <0.8, http-types >=0.12 && <0.13, jitsurei, - keiro ^>=0.22.0.0, - keiro-migrations ^>=0.22.0.0, - keiro-ops ^>=0.22.0.0, + keiro ^>=0.23.0.0, + keiro-migrations ^>=0.23.0.0, + keiro-ops ^>=0.23.0.0, keiro-ops-http, - keiro-pgmq ^>=0.22.0.0, - keiro-test-support ^>=0.22.0.0, - kiroku-metrics ^>=0.4.0.0, - kiroku-store ^>=0.12.0.0, + keiro-pgmq ^>=0.23.0.0, + keiro-test-support ^>=0.23.0.0, + kiroku-metrics ^>=0.4.0.1, + kiroku-store ^>=0.13.0.1, network >=3.1 && <3.3, optparse-applicative >=0.19 && <0.20, pgmq-migration >=0.7.0.0 && <0.8, diff --git a/src/Keiro/Ops/Http/Server.hs b/src/Keiro/Ops/Http/Server.hs --- a/src/Keiro/Ops/Http/Server.hs +++ b/src/Keiro/Ops/Http/Server.hs @@ -9,6 +9,7 @@ import Control.Concurrent.Async (Async, async, asyncThreadId, asyncWithUnmask, cancel, link) import Control.Concurrent.MVar import Control.Exception (bracket, bracketOnError, finally, mask, onException) +import Data.IORef (atomicModifyIORef', newIORef, writeIORef) import Data.Map.Strict qualified as Map import Data.Text qualified as Text import Keiro.Ops.Http.Config @@ -31,22 +32,35 @@ actualPort <- fromIntegral <$> Socket.socketPort socket ready <- newEmptyMVar clients <- newMVar Map.empty + -- The accept loop hands one socket at a time to spawn. Retain ownership + -- even before Warp has constructed and started its connection action. + pending <- newIORef Nothing + let takePending = atomicModifyIORef' pending (\accepted -> (Nothing, accepted)) + accept listener = mask $ \_ -> do + accepted@(connection, _) <- Socket.accept listener + writeIORef pending (Just connection) + pure accepted let spawn :: ((forall a. IO a -> IO a) -> IO ()) -> IO () spawn action = mask $ \_ -> do + connection <- takePending registered <- newEmptyMVar client <- asyncWithUnmask $ \unmask -> - -- Warp must install its accepted-socket cleanup before waiting - -- for registration. Its first unmask occurs inside that bracket. - action (\io -> readMVar registered >> unmask io) `finally` do - tid <- readMVar registered - modifyMVar_ clients (\connections -> pure $! Map.delete tid connections) + -- Warp's thread manager may skip the action when shutdown wins + -- the startup race. Close our socket even if its cleanup never + -- gets installed. Socket.close is safe after Warp has closed it. + action (\io -> readMVar registered >> unmask io) + `finally` ( mapM_ Socket.close connection `finally` do + tid <- readMVar registered + modifyMVar_ clients (\connections -> pure $! Map.delete tid connections) + ) let tid = asyncThreadId client (modifyMVar_ clients (\connections -> pure $! Map.insert tid client connections) >> putMVar registered tid) `onException` (tryPutMVar registered tid >> cancel client) stopClients = do live <- modifyMVar clients (\connections -> pure (Map.empty, Map.elems connections)) mapM_ cancel live - settings = Warp.setFork spawn (Warp.setBeforeMainLoop (putMVar ready ()) (Warp.setTimeout config.requestTimeoutSeconds Warp.defaultSettings)) + takePending >>= mapM_ Socket.close + settings = Warp.setAccept accept $ Warp.setFork spawn (Warp.setBeforeMainLoop (putMVar ready ()) (Warp.setTimeout config.requestTimeoutSeconds Warp.defaultSettings)) worker <- async (restore (Warp.runSettingsSocket settings socket app) `finally` (Socket.close socket >> stopClients)) `onException` Socket.close socket (link worker >> restore (takeMVar ready)) `onException` (cancel worker >> Socket.close socket) pure (OpsHttpServer worker actualPort) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -586,7 +586,10 @@ Nothing -> expectationFailure "expired Warp connection remained open" it "joins idle accepted connections across repeated start/stop churn" $ \store -> do app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store - forM_ [1 .. (20 :: Int)] $ \_ -> + -- Connect returning does not mean the worker has started. Stop + -- immediately to exercise both queued and accepted connections, + -- including workers whose Warp body is skipped during shutdown. + forM_ [1 .. (100 :: Int)] $ \iteration -> withOpsHttpServer defaultOpsHttpServerConfig {port = 0} app $ \server -> bracket (Socket.socket Socket.AF_INET Socket.Stream Socket.defaultProtocol) Socket.close $ \client -> do Socket.connect client (Socket.SockAddrInet (fromIntegral server.serverPort) (Socket.tupleToHostAddress (127, 0, 0, 1))) @@ -595,7 +598,7 @@ case result of Just (Left (_ :: IOException)) -> pure () Just (Right bytes) | Char8.null bytes -> pure () - _ -> expectationFailure "stopped runner retained an idle connection" + _ -> expectationFailure ("stopped runner retained an idle connection on iteration " <> show iteration) around (withFreshDatabase fixture) $ do describe "process-manager discovery database boundary" $ do it "performs only aggregate-types verifier work and rejects parsing before borrowing" $ \url ->