When Programming is Not Fun When the iteration loop is long, programming becomes frustrating. Let’s say you have to change a low-level system that lacks a way to test it directly. To test the new code, you have to manually execute a complex series of time-consuming and onerous steps. When something goes wrong, all you know is either your code is broken (for some unknown reason), or you exposed an existing issue in the monstrous system that depends on your change. It’s even worse when the code cannot be tested in a reproducible way, and you get a different behavior with each run. If that isn’t frustrating enough, imagine you are unable to implement the solution you envisioned. You can’t directly access the low-level system because a series of components — meant to abstract, simplify and create a safe alternative to your low-level system — are in the way. The components accomplish little of their stated purposes, but force your final design to be slower and more complex than necessary. Finally, you learn that the dev environment you are testing doesn’t match the behavior of production in crucial, important ways. This is what feels like to write queries for many websites. It’s not fun. Let’s fix that. database In this post I’ll show how to make database fun by using packages that make testing easy. I’ll walk through a example of building a durable queue backed with PostgreSQL. programming Isolated Temporary Databases To make database programming fun, we need a quick way to iterate. We need to bring up instances of our database, PostgreSQL in my case, and directly test the queries we are writing. One excellent option is . I’m going to utilize a Haskell analog I wrote called and the test helper based on it: . [pg_tmp](http://ephemeralpg.org/) [tmp-postgres](http://hackage.haskell.org/package/tmp-postgres) [hspec-pg-transact](http://hackage.haskell.org/package/hspec-pg-transact) provides a helper for setting up a test suite that creates a temporary process. Tests can use the optional , which adds a test and runs the db transaction used for the test. [hspec-pg-transact](http://hackage.haskell.org/package/hspec-pg-transact) postgres [itDB](https://hackage.haskell.org/package/hspec-pg-transact-0.1.0.0/docs/Test-Hspec-DB.html#v:itDB) Here is a simple example of testing db queries with a temporary database. migrate "Query” $ "work" $ do [ |INSERT INTO thingsVALUES (‘me’) |] [sql|SELECT nameFROM things |]` ` [Only "me"] describeDB itDB execute_ sql query_ shouldReturn (You need to make sure that is on the . This is case for on macOS, but for Ubuntu you must add to the ) initdb PATH brew /usr/lib/postgresql/VERSION/bin PATH The full code, including import statements, is . Let’s walk through a larger example. here Making postgresql-simple-queue In one of my previous posts, I discussed an in . The problem with the queue, as , is that it’s not . However, PostgreSQL is durable, so let’s use it to make our queue as Cale did. memory queue for sending emails Cale Gibbard pointed out durable We’ll add new payloads with . will grab the latest payload for us to process. We’ll call on the payload once the processing is finished: enqueueDB tryLockDB dequeueDB data Payload = Payload{ pId :: PayloadId, pValue :: Value, pState :: State, pCreatedAt :: UTCTime, pModifiedAt :: UTCTime} deriving (Show, Eq) enqueueDB :: Value -> DB PayloadIdenqueueDB value = dopid <- liftIO randomIOexecute[sql| INSERT INTO payloads(id, value) VALUES (?, ?)|](pid, value)return $ PayloadId pid tryLockDB :: DB (Maybe Payload)tryLockDB = listToMaybe <$> query_[sql| UPDATE payloadsSET state='locked'WHERE id in( SELECT idFROM payloadsWHERE state='enqueued'ORDER BY created_at ASCLIMIT 1)RETURNING id, value, state,created_at,modified_at|] dequeueDB :: PayloadId -> DB ()dequeueDB payloadId = void $ execute[sql| UPDATE payloadsSET state='dequeued'WHERE id=?|]payloadId All the operations occur in a monad so we can compose them with other db operations and have them run in the same transaction. For instance, we could use this library to and create a customer in a single transaction. [DB](http://hackage.haskell.org/package/pg-transact-0.1.0.0/docs/Database-PostgreSQL-Transact.html#t:DB) enqueueDB Back to Testing We can utilize to write our tests: [hspec-pg-transact](http://hackage.haskell.org/package/hspec-pg-transact) main :: IO ()main = hspec spec spec :: Specspec = describeDB init "Queue" $ doitDB "empty gives nothing" $ dotryLockDB`shouldReturn` Nothing itDB "enqueue/lock/dequeue" $ dotheId <- enqueueDB $ String "!"Just Payload {..} <- tryLockDB pId \`shouldBe\` theId pValue \`shouldBe\` String "!" tryLockDB \`shouldReturn\` Nothing dequeueDB pId \`shouldReturn\` () tryLockDB \`shouldReturn\` Nothing We can load up our single test file, , and run it in isolation: [test/Database/QueueSpec.hs](https://github.com/jfischoff/postgresql-queue/blob/master/test/Database/PostgreSQL/Simple/QueueSpec.hs) > :l test/Database/QueueSpec.hs> :main --match=locksDatabase.Queuepostgresql:///test?host=/tmp/tmp-postgres26345&port=57473 empty locks nothingenqueue/lock/dequeueShutting Down Finished in 1.8647 seconds2 examples, 0 failures Two seconds isn’t considered great in the unit-testing world, but for database testing we are in good shape. Not only are the tests fast, but we can also quickly iterate by reloading and rerunning a single file, all in single-digit seconds. This actually scales quite well. The queries themselves take milliseconds; almost all of the time is taken for the temporary db startup, which we only need to do once. Stress Tests Simple tests are easy, but let’s write a more complicated stress test. Unlike this simple test, the stress test will need to run multiple transactions, so we can’t use the test helper. itDB In this test we create consumer threads that read from the queue and threads that messages: 10 1000 enqueue it "stress queue" $ \testDB -> dolet withPool' = withPool testDBelemCount = 1000 :: Intexpected = [0 .. elemCount-1] ref <- newIORef [] -- Make 10 queue consumer threadsloopThreads <- replicateM 10$ async$ fix $ \next -> dompayload <- withPool' tryLockcase mpayload ofNothing -> nextJust Payload {..} -> dolast <-atomicModifyIORef ref $\xs -> ( pValue : xs, length xs + 1)withPool' $ \conn ->dequeue conn pIdwhen (last < elemCount)next -- Fork a 1000 threads and-- enqueue an indexforM_ [0 .. elementCount - 1] $\i -> forkIO $ void $withPool' $ flip enqueue $toJSON i waitAnyCancel loopThreadsJust decoded <-mapM (decode . encode)<$> readIORef refsort decoded`shouldBe` sort expected This test works, but makes clear that there is an issue with our design. We have to poll the db constantly to see if there is new data. Going Full Postgres Luckily PostgreSQL has a solution for us, as : and . We make a new function, , to utilize PostgreSQL’s pubsub feature: lpsimth pointed out [NOTIFY](https://www.postgresql.org/docs/current/static/sql-notify.html) [LISTEN](https://www.postgresql.org/docs/current/static/sql-listen.html) lock notifyPayload :: Connection-> IO ()notifyPayload conn = doNotification {..} <-getNotification connunless (notificationChannel== "enqueue") $notifyPayload conn lock :: Connection-> IO Payloadlock conn = bracket_(Simple.execute_conn "LISTEN enqueue")(Simple.execute_conn "UNLISTEN enqueue")$ fix $ \continue -> dom <- tryLock conncase m ofNothing -> donotifyPayload conncontinueJust x -> return x We extend to also our channel after it enqueues a payload enqueueDB NOTIFY enqueue enqueueDB :: Value -> DB PayloadIdenqueueDB value = dopid <- liftIO randomIOexecute[sql| INSERT INTO payloads(id, value) VALUES (?, ?)NOTIFY enqueue;|](pid, value)return $ PayloadId pid We can write a similar test for , and our loop becomes: lock loopThreads <- replicateM 10 $async $ fix $ \next -> do-- blocks waiting for-- new payloadx <- withPool testDB locklastCount <-atomicModifyIORef ref $\xs -> ( pValue x : xs, length xs + 1)withPool testDB $ \conn ->dequeue conn (pId x)when (last < elemCount)next Much nicer (ignoring the super small margins for mobile formatting :p). The final queue is packaged up as on and . postgresql-simple-queue hackage github Fun is Hard to Write We went through a bunch of intermediate states before I pushed it to Hackage. I started with something simple, wrote stress tests, which thankfully found bugs, and added pubsub. Most incredibly, it was fun and not frustrating. Don’t take my word for it: try it. and are on Hackage/Stackage. [pg-transact](https://hackage.haskell.org/package/pg-transact) [hspec-pg-transact](https://hackage.haskell.org/package/hspec-pg-transact) Extra Credit: Email Queue Take 2 We can now write a better email process queue, discussed in my other post. First, we need a processing loop: forever $ dopayload <- withResourceconnectionPool locksendEmail payloadwithResource connectionPool $\conn -> dequeue conn(pId payload) ’s, provides this consumer loop for us and also does command line argument processing. We can make a queue consumer executable easily, like so: postgres-simple-queue [defaultMain](https://github.com/jfischoff/postgresql-queue/blob/master/src/Database/PostgreSQL/Simple/Queue/Main.hs#L133) main :: IO ()main = doenv <- newEnv DiscoverrunResourceT $ runAWS env $ defaultMain "aws-email-queue-consumer"$ \payload _ -> docase fromJSON $ pValue payload ofSuccess email -> doresp <- AWS.send $ makeEmail emaillogFailedRequest respError x -> throwIO $ userError$ "Failed to decode payload as an Email: "++ show x See for the full example. [EmailQueue](https://github.com/jfischoff/postgresql-queue/blob/master/examples/EmailQueue.hs)
