4.5 KiB
Simplex.Messaging.Server.MsgStore.Postgres
PostgreSQL message store: router-side stored procedures for message operations, COPY protocol for bulk import.
Source: Postgres.hs
MsgQueue is unit type
type MsgQueue PostgresMsgStore = (). There is no message queue object for Postgres — all message operations go directly to the database via stored procedures. Functions like getMsgQueue return pure ().
Partial interface — error stubs
Multiple MsgStoreClass methods are error "X not used": withActiveMsgQueues, unsafeWithAllMsgQueues, logQueueStates, withIdleMsgQueue, getQueueMessages_, tryDeleteMsg_, setOverQuota_, getQueueSize_, unsafeRunStore. These are required by the type class but not applicable to Postgres. Calling any at runtime crashes. Postgres overrides the default implementations of tryPeekMsg, tryDelMsg, tryDelPeekMsg, deleteExpiredMsgs, and getQueueSize with direct database calls.
writeMsg — quota logic in stored procedure
write_message(?,?,?,?,?,?,?) is a PostgreSQL stored procedure that returns (quota_written, was_empty). Quota enforcement happens in SQL, not in Haskell. This means quota logic is duplicated: STM store checks canWrite flag in Haskell, Postgres store checks in the database function. The two implementations must agree on quota semantics.
tryDelPeekMsg — variable row count
The stored procedure try_del_peek_msg returns 0, 1, or 2 rows. For the 1-row case, the code checks whether the returned message's messageId matches the requested msgId to distinguish "deleted, no next message" from "delete failed, current message returned." This disambiguation is possible because the stored procedure always returns available messages even when deletion doesn't match.
uninterruptibleMask_ on all database operations
All write operations (writeMsg, tryDelMsg, tryDelPeekMsg, deleteExpiredMsgs) and isolateQueue are wrapped in E.uninterruptibleMask_. This prevents async exceptions (e.g., client disconnect) from interrupting mid-transaction, which could leave database connections in an inconsistent state.
batchInsertMessages — COPY protocol
Uses PostgreSQL's COPY FROM STDIN protocol (DB.copy_ + DB.putCopyData + DB.putCopyEnd) for bulk message import, which is much faster than individual INSERTs. Messages are encoded to CSV format. Parse errors on individual records are logged and skipped — the import is error-tolerant. The entire operation runs in a single transaction (withTransaction).
exportDbMessages — batched I/O
Accumulates rows in an IORef list (prepended for O(1) insert), flushing every 1000 records with reverse to restore order. Uses DB.foldWithOptions_ with fetchQuantity = Fixed 1000 to avoid loading all messages into memory.
updateQueueCounts — two-step reset
Creates a temp table with aggregated message stats, then updates msg_queues in two steps: first zeros all queue counts, then applies actual stats from the temp table. The two-step approach handles queues with zero messages: they're reset by the first UPDATE but not touched by the second (no matching row in temp table).
toMessage — nanosecond precision lost
MkSystemTime ts 0 constructs timestamps with zero nanoseconds. Only whole seconds are stored in the database. Messages read from Postgres have coarser timestamps than messages in STM/Journal stores.
isolateQueue IS the transaction boundary
isolateQueue for Postgres does uninterruptibleMask_ $ withDB' op ... $ runReaderT a . DBTransaction. Each isolateQueue call creates a fresh DBTransaction carrying the DB connection. This is how tryPeekMsg_ (which uses asks dbConn) gets its connection. The withQueueLock is identity for Postgres, so isolateQueue provides no mutual exclusion — only the DB transaction provides isolation.
newMsgStore hardcodes useCache = False
newQueueStore @PostgresQueue (queueStoreCfg config, False) — the Postgres message store always disables queue caching. All lookups go directly to the database. Contrast with the Journal+Postgres combination where caching is enabled.
deleteQueueSize — size before delete
deleteQueueSize calls getQueueSize BEFORE deleteStoreQueue. The returned size is the count at query time — a concurrent writeMsg between the size query and the delete means the reported size is stale. This is acceptable because the size is used for statistics, not for correctness.