@@ -108,35 +108,42 @@ spec = do
108108 |> Task. fromIOEither
109109 |> Task. mapError toText
110110
111- let entityName = EntityName " ConnReleaseRegressionEntity"
112- let callback _event = Task. yield unit
113-
114- -- Warm-up: one subscribe/unsubscribe cycle forces the Hasql pool and
115- -- the store's listener connections to fully establish, so the baseline
116- -- reflects steady state rather than a lazily-growing pool. (The pooled
117- -- connection borrowed by withConnectionAndError is what would otherwise
118- -- inflate the post-loop count and mask the per-stream release.)
119- subscribeUnsubscribeCycles store entityName callback 1
120- AsyncTask. sleep 500 |> Task. mapError (\ _ -> " warm-up settle sleep failed"
121- baseline <- countUserBackends adminConn
122-
123- -- Now run N more cycles. Each opens ONE dedicated per-stream connection
124- -- that unsubscribe must release; with the leak the count would climb by
125- -- N. With the fix the dedicated connection is gone each cycle, so the
126- -- steady-state count does not grow.
127- let cycleN = 5 :: Int
128- subscribeUnsubscribeCycles store entityName callback cycleN
129-
130- -- Allow libpq teardown to settle so pg_stat_activity reflects the release.
131- AsyncTask. sleep 500 |> Task. mapError (\ _ -> " settle sleep failed"
132- afterCycles <- countUserBackends adminConn
133-
134- Hasql. release adminConn |> Task. fromIO
135- store. close |> discard
136-
137- -- No net growth across the N measured cycles: a leak would leave
138- -- baseline + cycleN dedicated connections; the fix keeps it at baseline.
139- (afterCycles <= baseline) |> shouldBe True
111+ -- Guaranteed teardown: release the admin connection and close the store
112+ -- whether the body succeeds, throws, or the final assertion fails.
113+ -- Otherwise a failure here leaks connections into later Postgres-gated
114+ -- specs and contaminates them.
115+ let cleanup = do
116+ Hasql. release adminConn |> Task. fromIO
117+ store. close |> Task. mapError toText |> Task. ignoreError
118+
119+ Task. finally cleanup do
120+ let entityName = EntityName " ConnReleaseRegressionEntity"
121+ let callback _event = Task. yield unit
122+
123+ -- Warm-up: one subscribe/unsubscribe cycle forces the Hasql pool and
124+ -- the store's listener connections to fully establish, so the baseline
125+ -- reflects steady state rather than a lazily-growing pool. (The pooled
126+ -- connection borrowed by withConnectionAndError is what would otherwise
127+ -- inflate the post-loop count and mask the per-stream release.)
128+ subscribeUnsubscribeCycles store entityName callback 1
129+ -- Poll until the backend count settles instead of a fixed sleep, so
130+ -- the baseline reflects steady state regardless of teardown timing.
131+ baseline <- pollSettledBackends adminConn Nothing
132+
133+ -- Now run N more cycles. Each opens ONE dedicated per-stream connection
134+ -- that unsubscribe must release; with the leak the count would climb by
135+ -- N. With the fix the dedicated connection is gone each cycle, so the
136+ -- steady-state count does not grow.
137+ let cycleN = 5 :: Int
138+ subscribeUnsubscribeCycles store entityName callback cycleN
139+
140+ -- Poll until the count settles to at-or-below the baseline (or a bounded
141+ -- timeout expires), so pg_stat_activity reflects the per-stream releases.
142+ afterCycles <- pollSettledBackends adminConn (Just baseline)
143+
144+ -- No net growth across the N measured cycles: a leak would leave
145+ -- baseline + cycleN dedicated connections; the fix keeps it at baseline.
146+ (afterCycles <= baseline) |> shouldBe True
140147
141148
142149
@@ -213,16 +220,46 @@ subscribeUnsubscribeCycles store entityName callback n =
213220 subscribeUnsubscribeCycles store entityName callback (n - 1 )
214221
215222
216- -- | Count the backends currently open for the test database user via
217- -- pg_stat_activity. The dedicated per-stream connections are opened by the same
218- -- user, so a leak shows up as a higher count.
223+ -- | Count the backends currently open for the test database user, scoped to the
224+ -- current database. The dedicated per-stream connections are opened by the same
225+ -- user against the same database, so a leak shows up as a higher count. Scoping
226+ -- to @current_database()@ keeps unrelated sessions for the same user on other
227+ -- databases from inflating the count.
219228countUserBackends :: Hasql. Connection-> Task Text Int64
220229countUserBackends conn = do
221- let query :: Text = " SELECT count(*) :: int8 FROM pg_stat_activity WHERE usename = current_user"
230+ let query :: Text = " SELECT count(*) :: int8 FROM pg_stat_activity WHERE usename = current_user AND datname = current_database() "
222231 let decoder = Decoders. singleRow (Decoders. column (Decoders. nonNullable Decoders. int8))
223232 let statement :: Statement Unit Int64 =
224233 HasqlStatement. Statement|> Text. toBytes |> Bytes. unwrap) Encoders. noParams decoder True
225234 result <- Session. run (Session. statement unit statement) conn |> Task. fromIO |> Task. map Result. fromEither
226235 case result of
227236 Err err -> Task. throw [fmt |pg_stat_activity count failed: #{err}|]
228237 Ok count -> Task. yield count
238+
239+
240+ -- | Poll 'countUserBackends' until it settles, replacing a fixed sleep so the
241+ -- test is deterministic regardless of how long libpq teardown takes.
242+ --
243+ -- "Settled" means two consecutive samples (50 ms apart) are equal AND, when a
244+ -- @target@ baseline is supplied, the count is at or below it. Polling stops as
245+ -- soon as it settles, or after a bounded number of attempts (≈5 s), returning
246+ -- the last sample either way so the caller's assertion still runs.
247+ pollSettledBackends :: Hasql. Connection-> Maybe Int64 -> Task Text Int64
248+ pollSettledBackends conn target = do
249+ let maxAttempts = 100 :: Int
250+ let stepMs = 50
251+ let atOrBelowTarget count =
252+ case target of
253+ Nothing -> True
254+ Just baseline -> count <= baseline
255+ let loop attempts previous = do
256+ current <- countUserBackends conn
257+ let settled = case previous of
258+ Just prev -> current == prev && atOrBelowTarget current
259+ Nothing -> False
260+ case settled || attempts >= maxAttempts of
261+ True -> Task. yield current
262+ False -> do
263+ AsyncTask. sleep stepMs |> Task. mapError (\ _ -> " poll settle sleep failed"
264+ loop (attempts + 1 ) (Just current)
265+ loop 0 Nothing
0 commit comments