diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -1,5 +1,33 @@
 # Revision history for streaming-postgresql-simple
 
+## 0.2.0.0  -- 2017-02-03
+
+### Correctly perform finalisation in `query` functions.
+
+The previous implementation would perform the necessary finalisation only if the
+stream was drained. Some handling was in-place such that exceptions wouldn't
+cause the stream to end prematurely, but this isn't enough. We now use 
+`MonadResource` to register an action to drain the stream.
+
+Users should now wrap calls using `query` with `runResourceT`:
+
+```haskell
+>>> runResourceT (S.mapM_ print (query c "SELECT * FROM t"))
+```
+
+### Correctly deal with transactions in `stream`
+
+`stream` requires a transaction in order to function. If there isn't a 
+transaction open, `stream` would create one, but if you manually called `commit`
+or `rollback` from within the stream, the internal state would become 
+inconsistent. This would lead to confusing error messages.
+
+We now watch the transaction state as we pull items out from the stream, and
+inform the user if the internal state is not what we expected. Further more,
+cleanup actions (commit/rolling back the transaction) now only happen if there
+is still a transaction open. 
+
+
 ## 0.1.0.0  -- 2017-02-02
 
 * First version. Released on an unsuspecting world. Mwahaha.
diff --git a/Database/PostgreSQL/Simple/Streaming.hs b/Database/PostgreSQL/Simple/Streaming.hs
--- a/Database/PostgreSQL/Simple/Streaming.hs
+++ b/Database/PostgreSQL/Simple/Streaming.hs
@@ -1,6 +1,9 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
-{-# LANGUAGE BangPatterns, OverloadedStrings, RecordWildCards,
-  ScopedTypeVariables #-}
+{-# LANGUAGE RecordWildCards #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 module Database.PostgreSQL.Simple.Streaming
   ( -- * Queries that return results
@@ -26,18 +29,20 @@
     -- * Streaming data in and out of PostgreSQL with @COPY@
   , copyIn
   , copyOut
+
+    -- * Re-exported symbols
+  , runResourceT
   ) where
 
 import Control.Exception.Safe
-       (MonadCatch, MonadMask, catch, catchAny, throwM, mask)
+       (MonadMask, catch, throwM, mask)
 import Control.Monad (unless)
 import Control.Monad.Catch (onException)
 import Control.Monad.IO.Class (MonadIO, liftIO)
-import Control.Monad.State (put)
 import Control.Monad.Trans.Class (lift)
 import Control.Monad.Trans.Reader (runReaderT)
 import Control.Monad.Trans.Resource
-       (MonadResource, register, unprotect)
+       (MonadResource, register, unprotect, runResourceT)
 import Control.Monad.Trans.State.Strict (runStateT)
 import qualified Data.ByteString as B
 import qualified Data.ByteString as B8
@@ -64,7 +69,7 @@
        (isFailedTransactionError, beginMode, commit)
 import Database.PostgreSQL.Simple.TypeInfo (getTypeInfo)
 import Database.PostgreSQL.Simple.Types (Query(..))
-import Streaming (Stream, unfold, distribute)
+import Streaming (Stream, unfold)
 import Streaming.Internal (Stream(..))
 import Streaming.Prelude (Of(..), reread, for, untilRight)
 import qualified Streaming.Prelude as S
@@ -112,40 +117,41 @@
 -}
 
 query
-  :: (ToRow q, FromRow r)
+  :: (ToRow q, FromRow r, MonadResource m)
   => Connection
   -> Query
   -> q
-  -> Stream (Of r) IO ()
+  -> Stream (Of r) m ()
 query conn template qs =
   doQuery fromRow conn template =<< liftIO (formatQuery conn template qs)
 
 -- | A version of 'query' that does not perform query substitution.
 query_
-  :: (FromRow r)
+  :: (FromRow r, MonadResource m)
   => Connection
   -> Query
-  -> Stream (Of r) IO ()
+  -> Stream (Of r) m ()
 query_ conn template@(Query que) = doQuery fromRow conn template que
 
 -- | A version of 'query' taking parser as argument.
 queryWith
-  :: (ToRow q)
+  :: (ToRow q, MonadResource m)
   => RowParser r
   -> Connection
   -> Query
   -> q
-  -> Stream (Of r) IO ()
+  -> Stream (Of r) m ()
 queryWith parser conn template qs =
   doQuery parser conn template =<< liftIO (formatQuery conn template qs)
 
 -- | A version of 'query_' taking parser as argument.
 queryWith_
-  :: RowParser r -> Connection -> Query -> Stream (Of r) IO ()
+  :: MonadResource m
+  => RowParser r -> Connection -> Query -> Stream (Of r) m ()
 queryWith_ parser conn template@(Query que) = doQuery parser conn template que
 
 doQuery
-  :: (MonadIO m, MonadCatch m)
+  :: (MonadResource m)
   => RowParser r -> Connection -> Query -> B.ByteString -> Stream (Of r) m ()
 doQuery parser conn q que = do
   ok <-
@@ -155,25 +161,42 @@
   unless ok $
     liftIO (withConnection conn LibPQ.errorMessage) >>=
     traverse_ (liftIO . throwM . flip QueryError q . C8.unpack)
-  (_, fin) <-
-    flip runStateT (return ()) $
-    distribute $
-    for (results conn) $ \result -> do
-      status <- liftIO (LibPQ.resultStatus result)
-      case status of
-        LibPQ.EmptyQuery -> put $ throwM $ QueryError "query: Empty query" q
-        LibPQ.CommandOk ->
-          put $ throwM $ QueryError "query resulted in a command response" q
-        LibPQ.CopyOut ->
-          put $ throwM $ QueryError "query: COPY TO is not supported" q
-        LibPQ.CopyIn ->
-          put $ throwM $ QueryError "query: COPY FROM is not supported" q
-        LibPQ.BadResponse -> put $ throwResultError "query" result status
-        LibPQ.NonfatalError -> put $ throwResultError "query" result status
-        LibPQ.FatalError -> put $ throwResultError "query" result status
-        _ -> catchAny (streamResult conn parser result) (put . throwM)
-  liftIO fin
+  yieldResults `onTermination` drainRemainingResults
 
+  where
+
+    drainRemainingResults :: MonadIO m => m ()
+    drainRemainingResults =
+      S.effects (results conn)
+
+    yieldResults =
+      for (results conn) $ \result -> do
+        status <- liftIO (LibPQ.resultStatus result)
+        case status of
+          LibPQ.EmptyQuery ->
+            liftIO $ throwM $ QueryError "query: Empty query" q
+
+          LibPQ.CommandOk ->
+            liftIO $ throwM $ QueryError "query resulted in a command response" q
+
+          LibPQ.CopyOut ->
+            liftIO $ throwM $ QueryError "query: COPY TO is not supported" q
+
+          LibPQ.CopyIn ->
+            liftIO $ throwM $ QueryError "query: COPY FROM is not supported" q
+
+          LibPQ.BadResponse ->
+            liftIO (throwResultError "query" result status)
+
+          LibPQ.NonfatalError ->
+            liftIO (throwResultError "query" result status)
+
+          LibPQ.FatalError ->
+            liftIO (throwResultError "query" result status)
+
+          _ ->
+            streamResult conn parser result
+
 results :: MonadIO m => Connection -> Stream (Of LibPQ.Result) m ()
 results = reread (liftIO . flip withConnection LibPQ.getResult)
 
@@ -205,6 +228,16 @@
 -- transaction if needed.   The cursor is given a unique temporary name,
 -- so the consumer may itself call 'stream'.
 --
+-- Due to the dependency on transactions, you must ensure that `commit`
+-- or `rollback` aren't called on the connection used to form a stream.
+-- Doing so causes the stream cursor to be released, making it impossible
+-- to stream more results. If you do perform a commit or rollback,
+-- 'stream' will raise an exception indicating that a transaction
+-- was aborted.
+--
+-- If you performing transaction writes in a stream, consider instead
+-- using save points, which will nest correctly with 'stream'.
+--
 -- Exceptions that may be thrown:
 --
 -- * 'Database.PostgreSQL.Simple.FormatError': the query string could not be formatted correctly.
@@ -288,8 +321,8 @@
     case stat of
       LibPQ.TransIdle    ->
         bracket (liftIO (beginMode transactionMode conn))
-                (\_ -> liftIO (commit conn))
-                (\_ -> go `onException` liftIO (rollback conn))
+                (\_ -> ifInTransaction $ liftIO (commit conn))
+                (\_ -> go `onException` ifInTransaction (liftIO (rollback conn)))
       LibPQ.TransInTrans -> go
       LibPQ.TransActive  -> fail "foldWithOpts FIXME:  PQ.TransActive"
          -- This _shouldn't_ occur in the current incarnation of
@@ -304,12 +337,20 @@
       LibPQ.TransUnknown -> fail "foldWithOpts FIXME:  PQ.TransUnknown"
          -- Not sure what this means.
   where
+    ifInTransaction m = do
+      stat <- liftIO (withConnection conn LibPQ.transactionStatus)
+      case stat of
+        LibPQ.TransInTrans -> m
+        _ -> return ()
+
     declare = do
         name <- newTempName conn
         _ <- execute_ conn $ mconcat
                  [ "DECLARE ", name, " NO SCROLL CURSOR FOR ", q ]
         return name
-    close name = do
+
+    close name =
+      ifInTransaction $
         (execute_ conn ("CLOSE " <> name) >> return ()) `catch` \ex ->
             -- Don't throw exception if CLOSE failed because the transaction is
             -- aborted.  Otherwise, it will throw away the original error.
@@ -324,6 +365,12 @@
                  byteString " FROM " <>
                  byteString name)
             fetches = untilRight $ do
+
+              stat <- liftIO (withConnection conn LibPQ.transactionStatus)
+              case stat of
+                LibPQ.TransInTrans -> return ()
+                _ -> fail "Stream transaction prematurely aborted"
+
               result <- liftIO (exec conn fetchQ)
               status <- liftIO (LibPQ.resultStatus result)
               case status of
diff --git a/streaming-postgresql-simple.cabal b/streaming-postgresql-simple.cabal
--- a/streaming-postgresql-simple.cabal
+++ b/streaming-postgresql-simple.cabal
@@ -1,5 +1,5 @@
 name:                streaming-postgresql-simple
-version:             0.1.0.0
+version:             0.2.0.0
 synopsis:            Stream postgresql-query results using the streaming library
 description:         This package provides incremental streaming functions for the @postgresql-simple@ library.
 license:             BSD3
@@ -21,7 +21,6 @@
   build-depends:       base >=4.9 && <4.10
                      , bytestring >= 0.10.8.1 && < 0.11
                      , exceptions >= 0.8.3 && < 0.9
-                     , mtl >= 2.2.1 && < 2.3
                      , postgresql-libpq >= 0.9.2.0 && < 0.10
                      , postgresql-simple >=0.5 && <0.6
                      , resourcet >= 1.1.8.1 && < 1.2