diff --git a/plans/2026-10-23-server-stats-persistence-implementation-plan.md b/plans/2026-10-23-server-stats-persistence-implementation-plan.md new file mode 100644 index 000000000..f071efc7a --- /dev/null +++ b/plans/2026-10-23-server-stats-persistence-implementation-plan.md @@ -0,0 +1,1594 @@ +# SMP Server Stats Persistence Implementation Plan + +**Goal:** SMP server stats survive crashes: periodic saves to Postgres (Postgres mode) or an atomically written file (memory mode), with active-queue sets saved incrementally. + +**Architecture:** A new `Simplex.Messaging.Server.StatsStore` module loads and saves `ServerStatsData` through a file or a database backend. `PeriodStats` tracks hashes added since the last save so the database receives only new hashes. `Server.hs` restores on start, recomputes store counts, saves on a timer, around the daily rollover, and on shutdown, all under `savingLock`. + +**Tech Stack:** Haskell (GHC 9.6.3), postgresql-simple, hspec, cabal with `-fserver_postgres`. + +**Spec:** `plans/2026-10-23-server-stats-persistence-design.md` + +## Global Constraints + +- Build and test with `-fserver_postgres`; PostgreSQL must run on `localhost:5432` (see `contributing/PROJECT.md`). Use the same flags for every command so cabal does not rebuild between steps: `cabal build -fserver_postgres --ghc-options=-O0`, `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="..."`. +- Never wait with `sleep`/polling loops; run commands in the foreground with a timeout. +- Format changed Haskell with fourmolu (`fourmolu -i `); files with CPP (`Server.hs`, `Env/STM.hs`, `StatsStore.hs`, test files with CPP) are formatted by fragment, matching surrounding style. +- No new compiler warnings (library uses `-Weverything`); imports used only under `dbServerPostgres` go inside the `#if defined(dbServerPostgres)` block. +- Comments only where code cannot state it (ordering, invariants); no plan/spec references in code. +- Commit messages: `: `, max 50 chars, imperative, no body, no co-author. +- `server_stats_queues.period_set` values are fixed: 1 msg day, 2 msg week, 3 msg month, 4 ntf day, 5 ntf week, 6 ntf month. +- Column order of `server_stats` in the migration, `statsFields`, and the `FromRow StatsRow` instance must be identical. +- Default save interval 300 seconds, INI key `[STORE_LOG] stats_save_interval`. + +**User decisions (already made):** +- Postgres-mode servers store stats in a Postgres table; the existing file is imported once when the database has no stats row, then renamed to `.bak`; afterwards the database is the only source and corrections are done with SQL. +- One `BIGINT` column per persisted metric (not a text blob or JSONB). +- Active-queue sets are stored incrementally in a hash table (`server_stats_queues`), only new hashes inserted per save, rows of reset sets deleted. +- Memory-mode servers save the file periodically with temp file + rename and no longer rename it to `.bak` at startup. +- Scope: the metrics persisted in the file today, plus recomputing `qCount`/`msgCount`/`ntfCount` from the store on every start. +- Save interval is an INI option with default 300 seconds, active only when `log_stats = on`. + +--- + +## File Structure + +| File | Responsibility | +|---|---| +| `src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs` | New migration `20261023_server_stats` | +| `src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql` | Regenerated schema dump | +| `src/Simplex/Messaging/Server/Env/STM.hs` | `statsSaveInterval` config, `defStatsSaveInterval`, pending tracking switch in `newEnv` | +| `src/Simplex/Messaging/Server/Main.hs`, `Main/Init.hs` | INI option read and generated ini comment | +| `src/Simplex/Messaging/Server/Stats.hs` | Pending hashes in `PeriodStats`, `newServerStats` switch | +| `src/Simplex/Messaging/Server/StatsStore.hs` (new) | Load/save stats via file or database | +| `src/Simplex/Messaging/Server.hs` | Restore, periodic save thread, rollover and shutdown saves | +| `simplexmq.cabal` | New library module and test module | +| `tests/CoreTests/ServerStatsTests.hs` (new) | Pending tracking, file and database store tests | +| `tests/ServerTests.hs`, `tests/SMPClient.hs`, `tests/Test.hs` | Server-level tests, test config, registration | + +--- + +### Task 1: Add server_stats schema migration + +**Goal:** Postgres tables for persisted stats exist and the schema dump matches. + +**Files:** +- Modify: `src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs` (migration list at the top; definitions appended at the end of the file) +- Modify: `src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql` (regenerated by the schema dump test) + +**Acceptance Criteria:** +- [ ] `server_stats` has `server_stats_id SMALLINT PRIMARY KEY CHECK (server_stats_id = 1)`, `saved_at`, `from_time` and 84 counter columns, all `BIGINT NOT NULL` +- [ ] `server_stats_queues(period_set SMALLINT, queue_hash BIGINT, PRIMARY KEY (period_set, queue_hash))` +- [ ] Down migration drops both tables +- [ ] `/SMP server schema dump/` tests pass + +**Verify:** `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/SMP server schema dump/"` → 2 examples, 0 failures (second run, after the first run rewrote the dump) + +**Steps:** + +- [ ] **Step 1: Register the migration** + +In `serverSchemaMigrations`, after the `20260918_expire_messages` entry: + +```haskell + ("20260918_expire_messages", m20260918_expire_messages, Just down_m20260918_expire_messages), + ("20261023_server_stats", m20261023_server_stats, Just down_m20261023_server_stats) + ] +``` + +- [ ] **Step 2: Append migration definitions at the end of the file** + +```haskell +m20261023_server_stats :: Text +m20261023_server_stats = + [r| +CREATE TABLE server_stats( + server_stats_id SMALLINT NOT NULL PRIMARY KEY CHECK (server_stats_id = 1), + saved_at BIGINT NOT NULL, + from_time BIGINT NOT NULL, + q_created BIGINT NOT NULL, + q_secured BIGINT NOT NULL, + q_deleted_all BIGINT NOT NULL, + q_deleted_new BIGINT NOT NULL, + q_deleted_secured BIGINT NOT NULL, + q_deleted_all_b BIGINT NOT NULL, + q_blocked BIGINT NOT NULL, + q_sub BIGINT NOT NULL, + q_sub_all_b BIGINT NOT NULL, + q_sub_auth BIGINT NOT NULL, + q_sub_duplicate BIGINT NOT NULL, + q_sub_prohibited BIGINT NOT NULL, + q_sub_end BIGINT NOT NULL, + q_sub_end_b BIGINT NOT NULL, + ntf_created BIGINT NOT NULL, + ntf_new_created BIGINT NOT NULL, + ntf_deleted BIGINT NOT NULL, + ntf_deleted_b BIGINT NOT NULL, + ntf_sub BIGINT NOT NULL, + ntf_sub_b BIGINT NOT NULL, + ntf_sub_auth BIGINT NOT NULL, + ntf_sub_duplicate BIGINT NOT NULL, + msg_sent BIGINT NOT NULL, + msg_sent_auth BIGINT NOT NULL, + msg_sent_quota BIGINT NOT NULL, + msg_sent_large BIGINT NOT NULL, + msg_sent_block BIGINT NOT NULL, + msg_recv BIGINT NOT NULL, + msg_recv_get BIGINT NOT NULL, + msg_get BIGINT NOT NULL, + msg_get_no_msg BIGINT NOT NULL, + msg_get_auth BIGINT NOT NULL, + msg_get_duplicate BIGINT NOT NULL, + msg_get_prohibited BIGINT NOT NULL, + msg_expired BIGINT NOT NULL, + msg_sent_ntf BIGINT NOT NULL, + msg_recv_ntf BIGINT NOT NULL, + msg_ntfs BIGINT NOT NULL, + msg_ntfs_b BIGINT NOT NULL, + msg_ntf_no_sub BIGINT NOT NULL, + msg_ntf_lost BIGINT NOT NULL, + msg_ntf_expired BIGINT NOT NULL, + p_relays_requests BIGINT NOT NULL, + p_relays_successes BIGINT NOT NULL, + p_relays_errors_connect BIGINT NOT NULL, + p_relays_errors_compat BIGINT NOT NULL, + p_relays_errors_other BIGINT NOT NULL, + p_relays_own_requests BIGINT NOT NULL, + p_relays_own_successes BIGINT NOT NULL, + p_relays_own_errors_connect BIGINT NOT NULL, + p_relays_own_errors_compat BIGINT NOT NULL, + p_relays_own_errors_other BIGINT NOT NULL, + p_msg_fwds_requests BIGINT NOT NULL, + p_msg_fwds_successes BIGINT NOT NULL, + p_msg_fwds_errors_connect BIGINT NOT NULL, + p_msg_fwds_errors_compat BIGINT NOT NULL, + p_msg_fwds_errors_other BIGINT NOT NULL, + p_msg_fwds_own_requests BIGINT NOT NULL, + p_msg_fwds_own_successes BIGINT NOT NULL, + p_msg_fwds_own_errors_connect BIGINT NOT NULL, + p_msg_fwds_own_errors_compat BIGINT NOT NULL, + p_msg_fwds_own_errors_other BIGINT NOT NULL, + p_msg_fwds_recv BIGINT NOT NULL, + rcv_services_assoc_new BIGINT NOT NULL, + rcv_services_assoc_duplicate BIGINT NOT NULL, + rcv_services_assoc_updated BIGINT NOT NULL, + rcv_services_assoc_removed BIGINT NOT NULL, + rcv_services_sub_count BIGINT NOT NULL, + rcv_services_sub_duplicate BIGINT NOT NULL, + rcv_services_sub_queues BIGINT NOT NULL, + rcv_services_sub_end BIGINT NOT NULL, + ntf_services_assoc_new BIGINT NOT NULL, + ntf_services_assoc_duplicate BIGINT NOT NULL, + ntf_services_assoc_updated BIGINT NOT NULL, + ntf_services_assoc_removed BIGINT NOT NULL, + ntf_services_sub_count BIGINT NOT NULL, + ntf_services_sub_duplicate BIGINT NOT NULL, + ntf_services_sub_queues BIGINT NOT NULL, + ntf_services_sub_end BIGINT NOT NULL, + rslv_reqs BIGINT NOT NULL, + rslv_succ BIGINT NOT NULL, + rslv_not_found BIGINT NOT NULL, + rslv_resolver_errs BIGINT NOT NULL, + rslv_disabled BIGINT NOT NULL +); + +CREATE TABLE server_stats_queues( + period_set SMALLINT NOT NULL, + queue_hash BIGINT NOT NULL, + PRIMARY KEY (period_set, queue_hash) +); + |] + +down_m20261023_server_stats :: Text +down_m20261023_server_stats = + [r| +DROP TABLE server_stats_queues; + +DROP TABLE server_stats; + |] +``` + +- [ ] **Step 3: Run the schema dump test to regenerate the dump** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/SMP server schema dump/"` +Expected: "verify and overwrite schema dump" FAILS once (dump differs) and rewrites `server_schema.sql`; "verify schema down migrations" passes. + +- [ ] **Step 4: Run it again** + +Run the same command. +Expected: 2 examples, 0 failures. `git diff --stat` shows `server_schema.sql` with the two new tables and their primary keys only. + +- [ ] **Step 5: Commit** + +```bash +git add src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql +git commit -m "smp server: add server_stats tables" +``` + +```json:metadata +{"files": ["src/Simplex/Messaging/Server/QueueStore/Postgres/Migrations.hs", "src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql"], "verifyCommand": "cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match=\"/SMP server schema dump/\"", "acceptanceCriteria": ["server_stats singleton row with 84 counter columns", "server_stats_queues with composite PK", "down migration drops both tables", "schema dump tests pass"], "modelTier": "mechanical"} +``` + +--- + +### Task 2: Add stats_save_interval config option + +**Goal:** `ServerConfig.statsSaveInterval` is read from `[STORE_LOG] stats_save_interval` (default 300) and documented in the generated ini. + +**Files:** +- Modify: `src/Simplex/Messaging/Server/Env/STM.hs` (export list near `defNtfExpirationHours`; constants after `defaultNtfExpiration`; `ServerConfig` after `serverStatsBackupFile`) +- Modify: `src/Simplex/Messaging/Server/Main.hs:497` (config record) +- Modify: `src/Simplex/Messaging/Server/Main/Init.hs:106-108` (ini text) +- Modify: `tests/SMPClient.hs:266` (test config) + +**Acceptance Criteria:** +- [ ] `defStatsSaveInterval = 300` exported from `Env.STM` +- [ ] `statsSaveInterval` read with `readIniDefault defStatsSaveInterval "STORE_LOG" "stats_save_interval"` +- [ ] Generated ini contains `# stats_save_interval = 300` after `# prometheus_interval = 60` +- [ ] `/Server CLIs/` tests pass + +**Verify:** `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Server CLIs/"` → 0 failures + +**Steps:** + +- [ ] **Step 1: Config field and default in `Env/STM.hs`** + +Export list, after `defNtfExpirationHours,`: + +```haskell + defNtfExpirationHours, + defStatsSaveInterval, +``` + +After the `defaultNtfExpiration` definition: + +```haskell +defStatsSaveInterval :: Int64 +defStatsSaveInterval = 300 +``` + +In `ServerConfig`, after `serverStatsBackupFile`: + +```haskell + -- | file to save and restore stats + serverStatsBackupFile :: Maybe FilePath, + -- | interval to save stats, seconds + statsSaveInterval :: Int64, +``` + +- [ ] **Step 2: Read the option in `Main.hs`** + +After `serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",`: + +```haskell + serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log", + statsSaveInterval = readIniDefault defStatsSaveInterval "STORE_LOG" "stats_save_interval" ini, +``` + +- [ ] **Step 3: Document it in the generated ini (`Main/Init.hs`)** + +Replace: + +```haskell + <> "# Log interval for real-time Prometheus metrics\n\ + \# prometheus_interval = 60\n\n\ + \[AUTH]\n\ +``` + +with: + +```haskell + <> "# Log interval for real-time Prometheus metrics\n\ + \# prometheus_interval = 60\n\n" + <> "# Interval to save server statistics, seconds (requires log_stats)\n" + <> ("# stats_save_interval = " <> tshow defStatsSaveInterval <> "\n\n") + <> "[AUTH]\n\ +``` + +The rest of the `[AUTH]` literal stays unchanged. + +- [ ] **Step 4: Test config (`tests/SMPClient.hs`, `cfgMS`)** + +After `serverStatsBackupFile = Nothing,`: + +```haskell + serverStatsBackupFile = Nothing, + statsSaveInterval = defStatsSaveInterval, +``` + +`SMPClient.hs` already imports `Simplex.Messaging.Server.Env.STM` unqualified. + +- [ ] **Step 5: Build and run CLI tests** + +Run: `cabal build -fserver_postgres --ghc-options=-O0 all` +Expected: builds with no new warnings. +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Server CLIs/"` +Expected: 0 failures. + +- [ ] **Step 6: Commit** + +```bash +git add src/Simplex/Messaging/Server/Env/STM.hs src/Simplex/Messaging/Server/Main.hs src/Simplex/Messaging/Server/Main/Init.hs tests/SMPClient.hs +git commit -m "smp server: add stats_save_interval option" +``` + +```json:metadata +{"files": ["src/Simplex/Messaging/Server/Env/STM.hs", "src/Simplex/Messaging/Server/Main.hs", "src/Simplex/Messaging/Server/Main/Init.hs", "tests/SMPClient.hs"], "verifyCommand": "cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match=\"/Server CLIs/\"", "acceptanceCriteria": ["defStatsSaveInterval = 300 exported", "statsSaveInterval read from STORE_LOG", "generated ini documents the option", "CLI tests pass"], "modelTier": "mechanical"} +``` + +--- + +### Task 3: Track pending active-queue hashes + +**Goal:** `PeriodStats` optionally records hashes added and sets reset since the last save; SMP `ServerStats` enables it only for Postgres-mode servers with stats persistence. + +**Files:** +- Modify: `src/Simplex/Messaging/Server/Stats.hs` (export list; `newServerStats`; `PeriodStats`; `newPeriodStats`; `periodStatCounts`; `updatePeriodStats`) +- Modify: `src/Simplex/Messaging/Server/Env/STM.hs:548,574` (`newEnv`) +- Create: `tests/CoreTests/ServerStatsTests.hs` +- Modify: `tests/Test.hs` (import and `describe` in "Core tests"), `simplexmq.cabal` (test `other-modules`, after `CoreTests.SOCKSSettings`) + +**Acceptance Criteria:** +- [ ] `newPeriodStats` has no pending tracking (XFTP and NTF servers unchanged) +- [ ] `newPendingPeriodStats` records each hash once per period until taken +- [ ] Day reset always, week reset on Monday, month reset on the 1st set `pendingReset = True` and clear `pendingAdded` +- [ ] `restorePendingHashes` merges taken hashes back (union, reset flags OR-ed) +- [ ] `newEnv` enables tracking only when `serverStatsBackupFile` is set and the store is Postgres +- [ ] `/Core tests/Server stats tests/pending period stats/` pass + +**Verify:** `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` → 4 examples, 0 failures + +**Steps:** + +- [ ] **Step 1: Write the failing tests** + +Create `tests/CoreTests/ServerStatsTests.hs`: + +```haskell +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} + +module CoreTests.ServerStatsTests where + +import Data.Hashable (hash) +import Data.IORef +import Data.IntSet (IntSet) +import qualified Data.IntSet as IS +import Data.Maybe (isNothing) +import Data.Time.Calendar (fromGregorian) +import Data.Time.Clock (UTCTime (..)) +import Simplex.Messaging.Protocol (EntityId (..)) +import Simplex.Messaging.Server.Stats +import Test.Hspec hiding (fit, it) +import Util + +serverStatsTests :: Spec +serverStatsTests = + describe "pending period stats" $ do + it "should not track pending hashes when disabled" testNoPending + it "should track hashes added since the last take" testPendingAdded + it "should reset pending hashes with period sets" testPendingReset + it "should merge restored pending hashes" testMergePending + +q1, q2, q3, q4, q5 :: EntityId +q1 = EntityId "queue1" +q2 = EntityId "queue2" +q3 = EntityId "queue3" +q4 = EntityId "queue4" +q5 = EntityId "queue5" + +hashes :: [EntityId] -> IntSet +hashes = IS.fromList . map (hash . unEntityId) + +utcDay :: Integer -> Int -> Int -> UTCTime +utcDay y m d = UTCTime (fromGregorian y m d) 3600 + +-- 2026-06-10 is Wednesday, 2026-06-08 is Monday, 2026-06-01 is Monday and the 1st +wednesday, monday, monday1st :: UTCTime +wednesday = utcDay 2026 6 10 +monday = utcDay 2026 6 8 +monday1st = utcDay 2026 6 1 + +noPending :: PendingHashes +noPending = PendingHashes {pendingReset = False, pendingAdded = IS.empty} + +resetPending :: PendingHashes +resetPending = PendingHashes {pendingReset = True, pendingAdded = IS.empty} + +added :: [EntityId] -> PendingHashes +added qs = PendingHashes {pendingReset = False, pendingAdded = hashes qs} + +testNoPending :: IO () +testNoPending = do + ps <- newPeriodStats + isNothing (pendingPeriods ps) `shouldBe` True + updatePeriodStats ps q1 + readIORef (day ps) `shouldReturn` hashes [q1] + ServerStats {activeQueues, activeQueuesNtf} <- newServerStats False wednesday + isNothing (pendingPeriods activeQueues) `shouldBe` True + isNothing (pendingPeriods activeQueuesNtf) `shouldBe` True + +testPendingAdded :: IO () +testPendingAdded = do + ServerStats {activeQueues = ps} <- newServerStats True wednesday + Just PendingPeriods {pendingDay, pendingWeek, pendingMonth} <- pure $ pendingPeriods ps + mapM_ (updatePeriodStats ps) [q1, q2, q1] + takePendingHashes pendingDay `shouldReturn` added [q1, q2] + takePendingHashes pendingDay `shouldReturn` noPending + updatePeriodStats ps q1 + takePendingHashes pendingDay `shouldReturn` noPending + updatePeriodStats ps q3 + takePendingHashes pendingDay `shouldReturn` added [q3] + takePendingHashes pendingWeek `shouldReturn` added [q1, q2, q3] + takePendingHashes pendingMonth `shouldReturn` added [q1, q2, q3] + readIORef (day ps) `shouldReturn` hashes [q1, q2, q3] + +testPendingReset :: IO () +testPendingReset = do + ps <- newPendingPeriodStats + Just PendingPeriods {pendingDay, pendingWeek, pendingMonth} <- pure $ pendingPeriods ps + updatePeriodStats ps q1 + PeriodStatCounts {dayCount, weekCount, monthCount} <- periodStatCounts ps wednesday + (dayCount, weekCount, monthCount) `shouldBe` ("1", "", "") + takePendingHashes pendingDay `shouldReturn` resetPending + takePendingHashes pendingWeek `shouldReturn` added [q1] + takePendingHashes pendingMonth `shouldReturn` added [q1] + updatePeriodStats ps q2 + _ <- periodStatCounts ps monday + takePendingHashes pendingDay `shouldReturn` resetPending + takePendingHashes pendingWeek `shouldReturn` resetPending + takePendingHashes pendingMonth `shouldReturn` added [q2] + readIORef (month ps) `shouldReturn` hashes [q1, q2] + updatePeriodStats ps q3 + _ <- periodStatCounts ps monday1st + mapM takePendingHashes [pendingDay, pendingWeek, pendingMonth] `shouldReturn` [resetPending, resetPending, resetPending] + mapM readIORef [day ps, week ps, month ps] `shouldReturn` [IS.empty, IS.empty, IS.empty] + updatePeriodStats ps q4 + takePendingHashes pendingMonth `shouldReturn` added [q4] + +testMergePending :: IO () +testMergePending = do + mergePendingHashes (added [q1]) (added [q2]) `shouldBe` added [q1, q2] + mergePendingHashes resetPending (added [q2]) `shouldBe` PendingHashes {pendingReset = True, pendingAdded = hashes [q2]} + ref <- newIORef $ added [q5] + restorePendingHashes ref PendingHashes {pendingReset = True, pendingAdded = hashes [q3, q4]} + readIORef ref `shouldReturn` PendingHashes {pendingReset = True, pendingAdded = hashes [q3, q4, q5]} +``` + +Register it. In `tests/Test.hs` imports, after `import CoreTests.SOCKSSettings`: + +```haskell +import CoreTests.SOCKSSettings +import CoreTests.ServerStatsTests +``` + +In "Core tests", after `describe "Message store tests" msgStoreTests`: + +```haskell + describe "Message store tests" msgStoreTests + describe "Server stats tests" serverStatsTests +``` + +In `simplexmq.cabal` test-suite `other-modules`, after `CoreTests.SOCKSSettings`: + +``` + CoreTests.SOCKSSettings + CoreTests.ServerStatsTests +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` +Expected: compile error — `pendingPeriods`, `PendingPeriods`, `PendingHashes`, `takePendingHashes`, `newPendingPeriodStats`, `mergePendingHashes`, `restorePendingHashes` not in scope; `newServerStats` applied to too many arguments. + +- [ ] **Step 3: Implement pending tracking in `Stats.hs`** + +Export list, replace `newPeriodStats,` with: + +```haskell + newPeriodStats, + newPendingPeriodStats, + PendingPeriods (..), + PendingHashes (..), + takePendingHashes, + restorePendingHashes, + mergePendingHashes, +``` + +`newServerStats`: change the signature and the two `PeriodStats` allocations, and add a `where`: + +```haskell +newServerStats :: Bool -> UTCTime -> IO ServerStats +newServerStats trackPending ts = do +``` + +```haskell + activeQueues <- newActiveQueues +``` + +```haskell + activeQueuesNtf <- newActiveQueues +``` + +and after the final `pure ServerStats {...}` expression of `newServerStats`: + +```haskell + rslvStats + } + where + newActiveQueues = if trackPending then newPendingPeriodStats else newPeriodStats +``` + +Replace the `PeriodStats` type and `newPeriodStats`: + +```haskell +data PeriodStats = PeriodStats + { day :: IORef IntSet, + week :: IORef IntSet, + month :: IORef IntSet, + pendingPeriods :: Maybe PendingPeriods + } + +-- hashes added and sets reset since the last successful save +data PendingPeriods = PendingPeriods + { pendingDay :: IORef PendingHashes, + pendingWeek :: IORef PendingHashes, + pendingMonth :: IORef PendingHashes + } + +data PendingHashes = PendingHashes + { pendingReset :: Bool, + pendingAdded :: IntSet + } + deriving (Eq, Show) + +newPeriodStats :: IO PeriodStats +newPeriodStats = newPeriodStats_ Nothing + +newPendingPeriodStats :: IO PeriodStats +newPendingPeriodStats = do + pendingDay <- newIORef noPendingHashes + pendingWeek <- newIORef noPendingHashes + pendingMonth <- newIORef noPendingHashes + newPeriodStats_ $ Just PendingPeriods {pendingDay, pendingWeek, pendingMonth} + +newPeriodStats_ :: Maybe PendingPeriods -> IO PeriodStats +newPeriodStats_ pendingPeriods = do + day <- newIORef IS.empty + week <- newIORef IS.empty + month <- newIORef IS.empty + pure PeriodStats {day, week, month, pendingPeriods} + +noPendingHashes :: PendingHashes +noPendingHashes = PendingHashes {pendingReset = False, pendingAdded = IS.empty} + +takePendingHashes :: IORef PendingHashes -> IO PendingHashes +takePendingHashes ref = atomicModifyIORef' ref (noPendingHashes,) + +restorePendingHashes :: IORef PendingHashes -> PendingHashes -> IO () +restorePendingHashes ref taken = atomicModifyIORef'_ ref $ mergePendingHashes taken + +mergePendingHashes :: PendingHashes -> PendingHashes -> PendingHashes +mergePendingHashes (PendingHashes r hs) (PendingHashes r' hs') = PendingHashes (r || r') (IS.union hs hs') +``` + +Replace `periodStatCounts`: + +```haskell +periodStatCounts :: PeriodStats -> UTCTime -> IO PeriodStatCounts +periodStatCounts ps ts = do + let d = utctDay ts + (_, wDay) = mondayStartWeek d + MonthDay _ mDay = d + pending_ = pendingPeriods ps + dayCount <- periodCount 1 (day ps) (pendingDay <$> pending_) + weekCount <- periodCount wDay (week ps) (pendingWeek <$> pending_) + monthCount <- periodCount mDay (month ps) (pendingMonth <$> pending_) + pure PeriodStatCounts {dayCount, weekCount, monthCount} + where + periodCount :: Int -> IORef IntSet -> Maybe (IORef PendingHashes) -> IO Text + periodCount 1 ref pRef_ = do + -- pending is reset before the set, so a hash added concurrently can only be persisted for the new period + mapM_ (`atomicWriteIORef` PendingHashes {pendingReset = True, pendingAdded = IS.empty}) pRef_ + tshow . IS.size <$> atomicSwapIORef ref IS.empty + periodCount _ _ _ = pure "" +``` + +Replace `updatePeriodStats`: + +```haskell +updatePeriodStats :: PeriodStats -> EntityId -> IO () +updatePeriodStats ps (EntityId pId) = do + updatePeriod (day ps) (pendingDay <$> pendingPeriods ps) + updatePeriod (week ps) (pendingWeek <$> pendingPeriods ps) + updatePeriod (month ps) (pendingMonth <$> pendingPeriods ps) + where + ph = hash pId + updatePeriod ref pRef_ = unlessM (IS.member ph <$> readIORef ref) $ do + atomicModifyIORef'_ ref $ IS.insert ph + mapM_ (\pRef -> atomicModifyIORef'_ pRef $ \p -> p {pendingAdded = IS.insert ph $ pendingAdded p}) pRef_ +``` + +- [ ] **Step 4: Enable tracking in `newEnv` (`Env/STM.hs`)** + +Add `serverStatsBackupFile` to the `newEnv` pattern: + +```haskell +newEnv config@ServerConfig {smpCredentials, httpCredentials, serverStoreCfg, smpAgentCfg, information, messageExpiration, msgQueueQuota, namesConfig, msgIdBytes, serverStatsBackupFile} = do +``` + +Replace `serverStats <- newServerStats =<< getCurrentTime`: + +```haskell + serverStats <- newServerStats (isJust serverStatsBackupFile && isJust (dbStoreCfg serverStoreCfg)) =<< getCurrentTime +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` +Expected: 4 examples, 0 failures. `cabal build -fserver_postgres --ghc-options=-O0 all` shows no new warnings (XFTP and NTF servers compile unchanged). + +- [ ] **Step 6: Commit** + +```bash +git add src/Simplex/Messaging/Server/Stats.hs src/Simplex/Messaging/Server/Env/STM.hs tests/CoreTests/ServerStatsTests.hs tests/Test.hs simplexmq.cabal +git commit -m "smp server: track pending active queue hashes" +``` + +```json:metadata +{"files": ["src/Simplex/Messaging/Server/Stats.hs", "src/Simplex/Messaging/Server/Env/STM.hs", "tests/CoreTests/ServerStatsTests.hs", "tests/Test.hs", "simplexmq.cabal"], "verifyCommand": "cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match=\"/Core tests/Server stats tests/\"", "acceptanceCriteria": ["newPeriodStats without pending", "pending hashes recorded once per period", "reset flags per day/Monday/1st", "restore merges taken hashes", "newEnv tracks pending only for Postgres with persistence", "4 tests pass"], "modelTier": "standard"} +``` + +--- + +### Task 4: Add StatsStore module + +**Goal:** `loadStats`/`saveStats` persist `ServerStatsData` to a file (atomic replace) or to Postgres (row upsert, incremental hashes, one-time file import). + +**Files:** +- Create: `src/Simplex/Messaging/Server/StatsStore.hs` +- Modify: `simplexmq.cabal` (library `exposed-modules` in the `!flag(client_library)` block, after `Simplex.Messaging.Server.Stats`) +- Modify: `tests/CoreTests/ServerStatsTests.hs` + +**Acceptance Criteria:** +- [ ] File save writes `.tmp` then renames; load does not rename; missing file gives `Nothing`; invalid file throws; write errors are logged, not thrown +- [ ] DB save deletes rows of flagged sets, inserts pending hashes (batches of 10000, `ON CONFLICT DO NOTHING`), upserts the row, in one transaction +- [ ] DB save failure logs and restores taken pending hashes +- [ ] DB load returns the row and all six sets; unknown `period_set` throws +- [ ] DB load with no row imports the file in one transaction and renames it to `.bak`; with a row, the file is left in place with a warning +- [ ] Round trip preserves every persisted field (distinct values per field) + +**Verify:** `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` → 13 examples, 0 failures + +**Steps:** + +- [ ] **Step 1: Write the failing tests** + +In `tests/CoreTests/ServerStatsTests.hs` add `{-# LANGUAGE CPP #-}` to the pragmas, extend imports: + +```haskell +import qualified Data.ByteString.Char8 as B +import Simplex.Messaging.Encoding.String +import Simplex.Messaging.Server.StatsStore +import System.Directory (doesFileExist) +#if defined(dbServerPostgres) +import Control.Exception (bracket) +import Control.Monad (void) +import CoreTests.MsgStoreTests (testPostgresStoreConfig) +import qualified Database.PostgreSQL.Simple as DB +import Data.String (fromString) +import Simplex.Messaging.Agent.Store.Postgres.Common (DBStore, withTransaction) +import Simplex.Messaging.Server.MsgStore.Postgres (PostgresMsgStore) +import Simplex.Messaging.Server.MsgStore.Types (MsgStoreClass (..)) +import Simplex.Messaging.Server.QueueStore.Postgres (PostgresQueueStore (..)) +#endif +``` + +Replace `serverStatsTests`: + +```haskell +serverStatsTests :: Spec +serverStatsTests = do + describe "pending period stats" $ do + it "should not track pending hashes when disabled" testNoPending + it "should track hashes added since the last take" testPendingAdded + it "should reset pending hashes with period sets" testPendingReset + it "should merge restored pending hashes" testMergePending + describe "stats file" $ do + it "should save and load stats" testFileSaveLoad + it "should return no stats without file" testFileMissing + it "should log and continue when file cannot be written" testFileSaveError + it "should fail on invalid file" testFileParseError +#if defined(dbServerPostgres) + describe "stats database" $ do + it "should save and load stats with period sets" testDBSaveLoad + it "should save added hashes and delete reset sets" testDBResetAndAdded + it "should keep pending hashes when save fails" testDBSaveFailure + it "should import stats file when database has no stats" testDBImportFile + it "should ignore stats file when database has stats" testDBIgnoreFile +#endif +``` + +Append the helpers and tests: + +```haskell +statsFile :: HasTestEnv => FilePath +statsFile = testPath "smp-server-stats.log" + +-- every zero counter line of the file encoding gets a distinct value, so a column order mistake changes the result +distinctCounters :: ServerStatsData -> ServerStatsData +distinctCounters d = either error id $ strDecode $ B.unlines $ zipWith setValue [1 :: Int ..] $ B.lines $ strEncode d + where + setValue i l = case B.breakSubstring "=" l of + (k, "=0") -> k <> "=" <> B.pack (show i) + _ -> l + +newDistinctStats :: Bool -> IO ServerStats +newDistinctStats trackPending = do + ss <- newServerStats trackPending wednesday + setServerStats ss . distinctCounters =<< getServerStatsData ss + pure ss + +testFileSaveLoad :: HasTestEnv => IO () +testFileSaveLoad = do + ss <- newDistinctStats False + updatePeriodStats (activeQueues ss) q1 + updatePeriodStats (activeQueuesNtf ss) q2 + saveStats (StatsFile statsFile) ss + doesFileExist (statsFile <> ".tmp") `shouldReturn` False + expected <- getServerStatsData ss + Just d <- loadStats (StatsFile statsFile) + strEncode d `shouldBe` strEncode expected + doesFileExist statsFile `shouldReturn` True + +testFileMissing :: HasTestEnv => IO () +testFileMissing = (isNothing <$> loadStats (StatsFile statsFile)) `shouldReturn` True + +testFileSaveError :: HasTestEnv => IO () +testFileSaveError = do + let f = testPath "missing/smp-server-stats.log" + saveStats (StatsFile f) =<< newServerStats False wednesday + doesFileExist f `shouldReturn` False + +testFileParseError :: HasTestEnv => IO () +testFileParseError = do + B.writeFile statsFile "invalid" + loadStats (StatsFile statsFile) `shouldThrow` anyIOException + +#if defined(dbServerPostgres) +withStatsDB :: HasTestEnv => (DBStore -> IO a) -> IO a +withStatsDB action = bracket (newMsgStore testPostgresStoreConfig :: IO PostgresMsgStore) closeMsgStore $ action . dbStore . queueStore + +testDBSaveLoad :: HasTestEnv => IO () +testDBSaveLoad = withStatsDB $ \st -> do + ss <- newDistinctStats True + mapM_ (updatePeriodStats $ activeQueues ss) [q1, q2] + updatePeriodStats (activeQueuesNtf ss) q3 + saveStats (StatsDB st statsFile) ss + expected <- getServerStatsData ss + Just d <- getDBStats st + strEncode d `shouldBe` strEncode expected {_qCount = 0} + _day (_activeQueues d) `shouldBe` hashes [q1, q2] + _month (_activeQueuesNtf d) `shouldBe` hashes [q3] + +testDBResetAndAdded :: HasTestEnv => IO () +testDBResetAndAdded = withStatsDB $ \st -> do + ss <- newServerStats True wednesday + let msgQueues = activeQueues ss + mapM_ (updatePeriodStats msgQueues) [q1, q2] + updatePeriodStats (activeQueuesNtf ss) q3 + saveStats (StatsDB st statsFile) ss + updatePeriodStats msgQueues q4 + _ <- periodStatCounts msgQueues wednesday + updatePeriodStats msgQueues q5 + saveStats (StatsDB st statsFile) ss + Just ServerStatsData {_activeQueues, _activeQueuesNtf} <- getDBStats st + _day _activeQueues `shouldBe` hashes [q5] + _week _activeQueues `shouldBe` hashes [q1, q2, q4, q5] + _month _activeQueues `shouldBe` hashes [q1, q2, q4, q5] + _day _activeQueuesNtf `shouldBe` hashes [q3] + +testDBSaveFailure :: HasTestEnv => IO () +testDBSaveFailure = withStatsDB $ \st -> do + ss <- newServerStats True wednesday + updatePeriodStats (activeQueues ss) q1 + saveStats (StatsDB st statsFile) ss + updatePeriodStats (activeQueues ss) q2 + renameTable st "server_stats_queues" "server_stats_queues_moved" + saveStats (StatsDB st statsFile) ss + renameTable st "server_stats_queues_moved" "server_stats_queues" + Just d <- getDBStats st + _day (_activeQueues d) `shouldBe` hashes [q1] + saveStats (StatsDB st statsFile) ss + Just d' <- getDBStats st + _day (_activeQueues d') `shouldBe` hashes [q1, q2] + where + renameTable st from to = withTransaction st $ \db -> void $ DB.execute_ db $ fromString $ "ALTER TABLE " <> from <> " RENAME TO " <> to + +testDBImportFile :: HasTestEnv => IO () +testDBImportFile = withStatsDB $ \st -> do + ss <- newDistinctStats False + updatePeriodStats (activeQueues ss) q1 + d <- getServerStatsData ss + B.writeFile statsFile $ strEncode d + Just imported <- loadStats (StatsDB st statsFile) + strEncode imported `shouldBe` strEncode d + doesFileExist statsFile `shouldReturn` False + doesFileExist (statsFile <> ".bak") `shouldReturn` True + Just saved <- getDBStats st + strEncode saved `shouldBe` strEncode d {_qCount = 0} + +testDBIgnoreFile :: HasTestEnv => IO () +testDBIgnoreFile = withStatsDB $ \st -> do + ss <- newServerStats True wednesday + updatePeriodStats (activeQueues ss) q1 + saveStats (StatsDB st statsFile) ss + saved <- getServerStatsData ss + B.writeFile statsFile . strEncode =<< getServerStatsData =<< newDistinctStats False + Just loaded <- loadStats (StatsDB st statsFile) + strEncode loaded `shouldBe` strEncode saved {_qCount = 0} + doesFileExist statsFile `shouldReturn` True +#endif +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` +Expected: compile error — module `Simplex.Messaging.Server.StatsStore` not found. + +- [ ] **Step 3: Create `src/Simplex/Messaging/Server/StatsStore.hs`** + +```haskell +{-# LANGUAGE CPP #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} +{-# LANGUAGE TypeApplications #-} + +module Simplex.Messaging.Server.StatsStore + ( StatsStore (..), + loadStats, + saveStats, +#if defined(dbServerPostgres) + PeriodSet (..), + getDBStats, +#endif + ) +where + +import qualified Control.Exception as E +import Control.Logger.Simple +import Control.Monad +import qualified Data.ByteString.Char8 as B +import qualified Data.Text as T +import Simplex.Messaging.Encoding.String +import Simplex.Messaging.Server.Stats +import Simplex.Messaging.Util (ifM, tshow) +import System.Directory (doesFileExist, removeFile, renameFile) +import UnliftIO.Exception (tryAny) +#if defined(dbServerPostgres) +import Data.Bifunctor (bimap) +import Data.IORef (IORef) +import Data.Int (Int16, Int64) +import Data.IntSet (IntSet) +import qualified Data.IntSet as IS +import Data.List (find, intersperse) +import qualified Data.List.NonEmpty as L +import qualified Data.Map.Strict as M +import Data.Maybe (catMaybes) +import Data.Time.Clock.POSIX (posixSecondsToUTCTime, utcTimeToPOSIXSeconds) +import Data.Time.Clock.System (SystemTime (..), getSystemTime) +import Database.PostgreSQL.Simple (Only (..), Query) +import qualified Database.PostgreSQL.Simple as DB +import Database.PostgreSQL.Simple.FromRow (FromRow (..), field) +import Database.PostgreSQL.Simple.ToField (ToField (..)) +import Simplex.Messaging.Agent.Store.Postgres.Common (DBStore, withTransaction) +import Simplex.Messaging.Util (toChunks, whenM) +#endif + +data StatsStore + = StatsFile FilePath +#if defined(dbServerPostgres) + | StatsDB DBStore FilePath +#endif + +loadStats :: StatsStore -> IO (Maybe ServerStatsData) +loadStats = \case + StatsFile f -> ifM (doesFileExist f) (Just <$> readStatsFile f) (pure Nothing) +#if defined(dbServerPostgres) + StatsDB st f -> + getDBStats st >>= \case + Just d -> do + whenM (doesFileExist f) $ logWarn $ "server stats file " <> T.pack f <> " is ignored, stats are restored from the database" + pure $ Just d + Nothing -> ifM (doesFileExist f) (Just <$> importStatsFile st f) (pure Nothing) +#endif + +saveStats :: StatsStore -> ServerStats -> IO () +saveStats = \case + StatsFile f -> saveStatsFile f +#if defined(dbServerPostgres) + StatsDB st _ -> saveDBStats st +#endif + +readStatsFile :: FilePath -> IO ServerStatsData +readStatsFile f = do + logNote $ "restoring server stats from file " <> T.pack f + strDecode <$> B.readFile f >>= \case + Right d -> pure d + Left e -> E.throwIO $ userError $ "error parsing server stats file " <> f <> ": " <> e + +saveStatsFile :: FilePath -> ServerStats -> IO () +saveStatsFile f ss = do + d <- getServerStatsData ss + tryAny (B.writeFile tmp (strEncode d) >> renameFile tmp f) >>= \case + Right () -> pure () + Left e -> do + logError $ "error saving server stats to file " <> T.pack f <> ": " <> tshow e + void $ tryAny $ removeFile tmp + where + tmp = f <> ".tmp" + +#if defined(dbServerPostgres) +data PeriodSet = PSMsgDay | PSMsgWeek | PSMsgMonth | PSNtfDay | PSNtfWeek | PSNtfMonth + deriving (Bounded, Enum, Eq, Ord, Show) + +-- stored in server_stats_queues.period_set, the values must not change +periodSetId :: PeriodSet -> Int16 +periodSetId = \case + PSMsgDay -> 1 + PSMsgWeek -> 2 + PSMsgMonth -> 3 + PSNtfDay -> 4 + PSNtfWeek -> 5 + PSNtfMonth -> 6 + +instance ToField PeriodSet where + toField = toField . periodSetId + +periodSetFromId :: Int16 -> Maybe PeriodSet +periodSetFromId i = find ((i ==) . periodSetId) [minBound .. maxBound] + +periodSetHashes :: ServerStatsData -> PeriodSet -> IntSet +periodSetHashes ServerStatsData {_activeQueues = q, _activeQueuesNtf = n} = \case + PSMsgDay -> _day q + PSMsgWeek -> _week q + PSMsgMonth -> _month q + PSNtfDay -> _day n + PSNtfWeek -> _week n + PSNtfMonth -> _month n + +periodSetPending :: ServerStats -> PeriodSet -> Maybe (IORef PendingHashes) +periodSetPending ServerStats {activeQueues, activeQueuesNtf} = \case + PSMsgDay -> pendingDay <$> pendingPeriods activeQueues + PSMsgWeek -> pendingWeek <$> pendingPeriods activeQueues + PSMsgMonth -> pendingMonth <$> pendingPeriods activeQueues + PSNtfDay -> pendingDay <$> pendingPeriods activeQueuesNtf + PSNtfWeek -> pendingWeek <$> pendingPeriods activeQueuesNtf + PSNtfMonth -> pendingMonth <$> pendingPeriods activeQueuesNtf + +getDBStats :: DBStore -> IO (Maybe ServerStatsData) +getDBStats st = + withTransaction st $ \db -> + DB.query_ db selectStatsQuery >>= \case + StatsRow d : _ -> Just . withPeriodSets d <$> DB.fold_ db "SELECT period_set, queue_hash FROM server_stats_queues" M.empty addHash + [] -> pure Nothing + where + addHash sets (psId :: Int16, h :: Int64) = case periodSetFromId psId of + Just ps -> pure $! M.alter (Just . maybe (IS.singleton h') (IS.insert h')) ps sets + Nothing -> E.throwIO $ userError $ "unknown period_set " <> show psId + where + h' = fromIntegral h + withPeriodSets d sets = + d + { _activeQueues = periodData PSMsgDay PSMsgWeek PSMsgMonth, + _activeQueuesNtf = periodData PSNtfDay PSNtfWeek PSNtfMonth + } + where + periodData dp wp mp = PeriodStatsData {_day = get dp, _week = get wp, _month = get mp} + get p = M.findWithDefault IS.empty p sets + +importStatsFile :: DBStore -> FilePath -> IO ServerStatsData +importStatsFile st f = do + d <- readStatsFile f + withTransaction st $ \db -> do + void $ DB.execute_ db "DELETE FROM server_stats_queues" + forM_ [minBound .. maxBound] $ \ps -> insertHashes db ps $ periodSetHashes d ps + upsertStats db d + renameFile f $ f <> ".bak" + logNote "server stats imported to the database" + pure d + +saveDBStats :: DBStore -> ServerStats -> IO () +saveDBStats st ss = do + taken <- fmap catMaybes $ forM [minBound .. maxBound] $ \ps -> + forM (periodSetPending ss ps) $ \ref -> (ps,ref,) <$> takePendingHashes ref + d <- getServerStatsData ss + tryAny (withTransaction st $ \db -> mapM_ (savePending db) taken >> upsertStats db d) >>= \case + Right () -> pure () + Left e -> do + logError $ "error saving server stats to the database: " <> tshow e + forM_ taken $ \(_, ref, p) -> restorePendingHashes ref p + where + savePending db (ps, _, PendingHashes {pendingReset, pendingAdded}) = do + when pendingReset $ void $ DB.execute db "DELETE FROM server_stats_queues WHERE period_set = ?" (Only ps) + insertHashes db ps pendingAdded + +insertHashes :: DB.Connection -> PeriodSet -> IntSet -> IO () +insertHashes db ps hs = + forM_ (toChunks hashInsertBatch $ IS.toList hs) $ \hs' -> + void $ + DB.executeMany + db + "INSERT INTO server_stats_queues (period_set, queue_hash) VALUES (?, ?) ON CONFLICT DO NOTHING" + (map (\h -> (ps, fromIntegral h :: Int64)) $ L.toList hs') + +-- rows per INSERT statement, bounds the query size of the one-time file import +hashInsertBatch :: Int +hashInsertBatch = 10000 + +upsertStats :: DB.Connection -> ServerStatsData -> IO () +upsertStats db d = do + savedAt <- systemSeconds <$> getSystemTime + void $ DB.execute db upsertStatsQuery $ savedAt : map (($ d) . snd) statsFields + +selectStatsQuery :: Query +selectStatsQuery = "SELECT " <> commaSep (map fst statsFields) <> " FROM server_stats WHERE server_stats_id = 1" + +upsertStatsQuery :: Query +upsertStatsQuery = + "INSERT INTO server_stats (server_stats_id, saved_at, " + <> commaSep cols + <> ") VALUES (1, ?, " + <> commaSep (map (const "?") cols) + <> ") ON CONFLICT (server_stats_id) DO UPDATE SET " + <> commaSep (map (\c -> c <> " = EXCLUDED." <> c) ("saved_at" : cols)) + where + cols = map fst statsFields + +commaSep :: [Query] -> Query +commaSep = mconcat . intersperse ", " + +-- the order must match the FromRow StatsRow instance +statsFields :: [(Query, ServerStatsData -> Int64)] +statsFields = + ("from_time", round . utcTimeToPOSIXSeconds . _fromTime) + : map + (fmap (fromIntegral .)) + ( [ ("q_created", _qCreated), + ("q_secured", _qSecured), + ("q_deleted_all", _qDeletedAll), + ("q_deleted_new", _qDeletedNew), + ("q_deleted_secured", _qDeletedSecured), + ("q_deleted_all_b", _qDeletedAllB), + ("q_blocked", _qBlocked), + ("q_sub", _qSub), + ("q_sub_all_b", _qSubAllB), + ("q_sub_auth", _qSubAuth), + ("q_sub_duplicate", _qSubDuplicate), + ("q_sub_prohibited", _qSubProhibited), + ("q_sub_end", _qSubEnd), + ("q_sub_end_b", _qSubEndB), + ("ntf_created", _ntfCreated), + ("ntf_new_created", _ntfNewCreated), + ("ntf_deleted", _ntfDeleted), + ("ntf_deleted_b", _ntfDeletedB), + ("ntf_sub", _ntfSub), + ("ntf_sub_b", _ntfSubB), + ("ntf_sub_auth", _ntfSubAuth), + ("ntf_sub_duplicate", _ntfSubDuplicate), + ("msg_sent", _msgSent), + ("msg_sent_auth", _msgSentAuth), + ("msg_sent_quota", _msgSentQuota), + ("msg_sent_large", _msgSentLarge), + ("msg_sent_block", _msgSentBlock), + ("msg_recv", _msgRecv), + ("msg_recv_get", _msgRecvGet), + ("msg_get", _msgGet), + ("msg_get_no_msg", _msgGetNoMsg), + ("msg_get_auth", _msgGetAuth), + ("msg_get_duplicate", _msgGetDuplicate), + ("msg_get_prohibited", _msgGetProhibited), + ("msg_expired", _msgExpired), + ("msg_sent_ntf", _msgSentNtf), + ("msg_recv_ntf", _msgRecvNtf), + ("msg_ntfs", _msgNtfs), + ("msg_ntfs_b", _msgNtfsB), + ("msg_ntf_no_sub", _msgNtfNoSub), + ("msg_ntf_lost", _msgNtfLost), + ("msg_ntf_expired", _msgNtfExpired) + ] + <> proxyFields "p_relays" _pRelays + <> proxyFields "p_relays_own" _pRelaysOwn + <> proxyFields "p_msg_fwds" _pMsgFwds + <> proxyFields "p_msg_fwds_own" _pMsgFwdsOwn + <> [("p_msg_fwds_recv", _pMsgFwdsRecv)] + <> serviceFields "rcv_services" _rcvServices + <> serviceFields "ntf_services" _ntfServices + <> map + (fmap (. _rslvStats)) + [ ("rslv_reqs", _rslvReqs), + ("rslv_succ", _rslvSucc), + ("rslv_not_found", _rslvNotFound), + ("rslv_resolver_errs", _rslvResolverErrs), + ("rslv_disabled", _rslvDisabled) + ] + ) + where + proxyFields pfx sel = + map + (bimap ((pfx <> "_") <>) (. sel)) + [ ("requests", _pRequests), + ("successes", _pSuccesses), + ("errors_connect", _pErrorsConnect), + ("errors_compat", _pErrorsCompat), + ("errors_other", _pErrorsOther) + ] + serviceFields pfx sel = + map + (bimap ((pfx <> "_") <>) (. sel)) + [ ("assoc_new", _srvAssocNew), + ("assoc_duplicate", _srvAssocDuplicate), + ("assoc_updated", _srvAssocUpdated), + ("assoc_removed", _srvAssocRemoved), + ("sub_count", _srvSubCount), + ("sub_duplicate", _srvSubDuplicate), + ("sub_queues", _srvSubQueues), + ("sub_end", _srvSubEnd) + ] + +newtype StatsRow = StatsRow ServerStatsData + +instance FromRow StatsRow where + fromRow = do + _fromTime <- posixSecondsToUTCTime . fromIntegral <$> field @Int64 + _qCreated <- field + _qSecured <- field + _qDeletedAll <- field + _qDeletedNew <- field + _qDeletedSecured <- field + _qDeletedAllB <- field + _qBlocked <- field + _qSub <- field + _qSubAllB <- field + _qSubAuth <- field + _qSubDuplicate <- field + _qSubProhibited <- field + _qSubEnd <- field + _qSubEndB <- field + _ntfCreated <- field + _ntfNewCreated <- field + _ntfDeleted <- field + _ntfDeletedB <- field + _ntfSub <- field + _ntfSubB <- field + _ntfSubAuth <- field + _ntfSubDuplicate <- field + _msgSent <- field + _msgSentAuth <- field + _msgSentQuota <- field + _msgSentLarge <- field + _msgSentBlock <- field + _msgRecv <- field + _msgRecvGet <- field + _msgGet <- field + _msgGetNoMsg <- field + _msgGetAuth <- field + _msgGetDuplicate <- field + _msgGetProhibited <- field + _msgExpired <- field + _msgSentNtf <- field + _msgRecvNtf <- field + _msgNtfs <- field + _msgNtfsB <- field + _msgNtfNoSub <- field + _msgNtfLost <- field + _msgNtfExpired <- field + _pRelays <- proxyStatsP + _pRelaysOwn <- proxyStatsP + _pMsgFwds <- proxyStatsP + _pMsgFwdsOwn <- proxyStatsP + _pMsgFwdsRecv <- field + _rcvServices <- serviceStatsP + _ntfServices <- serviceStatsP + _rslvStats <- NameResolverStatsData <$> field <*> field <*> field <*> field <*> field + pure $ + StatsRow + ServerStatsData + { _fromTime, + _qCreated, + _qSecured, + _qDeletedAll, + _qDeletedAllB, + _qDeletedNew, + _qDeletedSecured, + _qBlocked, + _qSub, + _qSubAllB, + _qSubAuth, + _qSubDuplicate, + _qSubProhibited, + _qSubEnd, + _qSubEndB, + _ntfCreated, + _ntfNewCreated, + _ntfDeleted, + _ntfDeletedB, + _ntfSub, + _ntfSubB, + _ntfSubAuth, + _ntfSubDuplicate, + _msgSent, + _msgSentAuth, + _msgSentQuota, + _msgSentLarge, + _msgSentBlock, + _msgRecv, + _msgRecvAckTimes = emptyTimeBuckets, + _msgRecvGet, + _msgGet, + _msgGetNoMsg, + _msgGetAuth, + _msgGetDuplicate, + _msgGetProhibited, + _msgExpired, + _activeQueues = newPeriodStatsData, + _msgSentNtf, + _msgRecvNtf, + _activeQueuesNtf = newPeriodStatsData, + _msgNtfs, + _msgNtfsB, + _msgNtfNoSub, + _msgNtfLost, + _msgNtfExpired, + _pRelays, + _pRelaysOwn, + _pMsgFwds, + _pMsgFwdsOwn, + _pMsgFwdsRecv, + _ntfServices, + _rcvServices, + _rcvServicesSubMsg = 0, + _rcvServicesSubDuplicate = 0, + _authDelayExceeded = 0, + _cmdSlotWaits = 0, + _cmdSlotWaitMs = 0, + _cmdSlotWaitMaxMs = 0, + _cmdSlotWaitTimeouts = 0, + _qCount = 0, + _msgCount = 0, + _ntfCount = 0, + _rslvStats + } + where + proxyStatsP = ProxyStatsData <$> field <*> field <*> field <*> field <*> field + serviceStatsP = do + _srvAssocNew <- field + _srvAssocDuplicate <- field + _srvAssocUpdated <- field + _srvAssocRemoved <- field + _srvSubCount <- field + _srvSubDuplicate <- field + _srvSubQueues <- field + _srvSubEnd <- field + pure newServiceStatsData {_srvAssocNew, _srvAssocDuplicate, _srvAssocUpdated, _srvAssocRemoved, _srvSubCount, _srvSubDuplicate, _srvSubQueues, _srvSubEnd} +#endif +``` + +In `simplexmq.cabal`, library `exposed-modules` (the `!flag(client_library)` block), after `Simplex.Messaging.Server.Stats`: + +``` + Simplex.Messaging.Server.Stats + Simplex.Messaging.Server.StatsStore +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` +Expected: 13 examples, 0 failures (the failing-save test logs one "error saving server stats to the database" line). +Run: `cabal build --ghc-options=-O0 lib:simplexmq` (without `-fserver_postgres`) +Expected: builds with no warnings from `StatsStore.hs`. + +- [ ] **Step 5: Commit** + +```bash +git add src/Simplex/Messaging/Server/StatsStore.hs simplexmq.cabal tests/CoreTests/ServerStatsTests.hs +git commit -m "smp server: add stats store" +``` + +```json:metadata +{"files": ["src/Simplex/Messaging/Server/StatsStore.hs", "simplexmq.cabal", "tests/CoreTests/ServerStatsTests.hs"], "verifyCommand": "cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match=\"/Core tests/Server stats tests/\"", "acceptanceCriteria": ["atomic file save, no rename on load", "DB save deletes reset sets, inserts pending, upserts row in one tx", "failed DB save restores pending", "DB load with sets; unknown period_set throws", "import once with .bak; row wins over file", "round trip with distinct values", "13 tests pass", "builds without server_postgres"], "modelTier": "frontier"} +``` + +--- + +### Task 5: Save stats periodically in the server + +**Goal:** The SMP server restores stats through `StatsStore`, recomputes store counts on every start, and saves stats on a timer, around the daily rollover and on shutdown under `savingLock`. + +**Files:** +- Modify: `src/Simplex/Messaging/Server.hs` (imports; `raceAny_` thread list ~:186; `saveServer` ~:239; `serverStatsThread_`/`logServerStats` ~:510-530; `saveServerStats`/`restoreServerStats` ~:2515-2549) +- Modify: `tests/ServerTests.hs` (imports; `serverTests` list; `testRestoreMessages`; `testRestoreExpireMessages`; new tests) + +**Acceptance Criteria:** +- [ ] Postgres mode loads from `server_stats` (or imports the file once); memory mode loads the file without renaming +- [ ] Load errors log and exit +- [ ] `qCount`, `msgCount`, `ntfCount` are set from the store on every start; startup expiration counts are added to `msgExpired`/`msgNtfExpired` +- [ ] `saveStatsThread` saves every `statsSaveInterval` seconds under `savingLock` when `serverStatsBackupFile` is set +- [ ] Daily rollover saves before and after writing the daily line, holding `savingLock` +- [ ] Shutdown and control `save` save stats before the message store is closed +- [ ] Existing restore tests pass for memory and Postgres stores; new tests pass + +**Verify:** `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/SMP server via TLS/"` → 0 failures + +**Steps:** + +- [ ] **Step 1: Write the failing server tests (`tests/ServerTests.hs`)** + +Imports: change `import Data.Maybe (catMaybes)` to `import Data.Maybe (catMaybes, fromMaybe)`. In the `#if defined(dbServerPostgres)` import block add: + +```haskell +import CoreTests.ServerStatsTests (withStatsDB) +import Simplex.Messaging.Server.StatsStore (getDBStats) +``` + +In `serverTests`, after `describe "Restore messages (old / v2)" testRestoreExpireMessages`: + +```haskell + describe "Restore messages (old / v2)" testRestoreExpireMessages + describe "Save server stats" $ do + testSaveStatsPeriodically + testStartupStoreCounts +``` + +Add after `checkStats`: + +```haskell +readSavedStats :: HasTestEnv => AStoreType -> IO ServerStatsData +readSavedStats = \case + ASType _ SMSMemory -> do + logSize testServerStatsBackupFile `shouldReturn` 101 + either error id . strDecode <$> B.readFile testServerStatsBackupFile +#if defined(dbServerPostgres) + ASType _ SMSPostgres -> withStatsDB $ fmap (fromMaybe $ error "no saved stats") . getDBStats +#else + ASType _ SMSPostgres -> error "compiled without server_postgres flag" +#endif + +testSaveStatsPeriodically :: SpecWith (ASrvTransport, AStoreType) +testSaveStatsPeriodically = + it "should save stats while running and restore them on start" $ \(at@(ATransport t), msType) -> do + removeFileIfExists testServerStatsBackupFile + g <- C.newRandom + (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + let cfg' = updateCfg (cfgMS msType) $ \cfg_ -> cfg_ {serverStatsBackupFile = Just testServerStatsBackupFile, statsSaveInterval = 1} + (rId, stats1, snapshot) <- withSmpServerConfigOn at cfg' testPort $ \_ -> do + rId <- runClient t $ \h -> runClient t $ \rh -> do + (sId, rId, _, _) <- createAndSecureQueue rh sPub + Resp "1" _ OK <- signSendRecv h sKey ("1", sId, _SEND "hello") + pure rId + threadDelay 1500000 + stats1 <- readSavedStats msType + snapshot <- case msType of + ASType _ SMSMemory -> Just <$> B.readFile testServerStatsBackupFile + _ -> pure Nothing + pure (rId, stats1, snapshot) + checkStats stats1 [rId] 1 0 + -- a crash leaves the last periodic save, not the save on shutdown + mapM_ (B.writeFile testServerStatsBackupFile) snapshot + stats2 <- withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1500000 >> readSavedStats msType + checkStats stats2 [rId] 1 0 + removeFileIfExists testServerStatsBackupFile + where + runClient :: (HasTestEnv, Transport c) => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO a) -> IO a + runClient _ = testSMPClient + +testStartupStoreCounts :: SpecWith (ASrvTransport, AStoreType) +testStartupStoreCounts = + it "should count stored queues and messages on start without saved stats" $ \(at@(ATransport t), msType) -> do + g <- C.newRandom + (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g + let cfg' = cfgMS msType + withSmpServerConfigOn at cfg' testPort $ \_ -> runClient t $ \h -> runClient t $ \rh -> do + (sId, _, _, _) <- createAndSecureQueue rh sPub + Resp "1" _ OK <- signSendRecv h sKey ("1", sId, _SEND "hello 1") + Resp "2" _ OK <- signSendRecv h sKey ("2", sId, _SEND "hello 2") + pure () + let cfg2 = updateCfg cfg' $ \cfg_ -> cfg_ {prometheusInterval = Just 1} + metrics <- withSmpServerConfigOn at cfg2 testPort $ \_ -> threadDelay 1500000 >> B.readFile testPrometheusMetricsFile + let expected = case msType of + ASType SQSMemory _ -> ["simplex_smp_messages_total 2", "simplex_smp_queues_total1 1"] + -- Postgres queue count is a reltuples estimate + _ -> ["simplex_smp_messages_total 2"] + forM_ expected $ \l -> B.lines metrics `shouldContain` [l] + where + runClient :: (HasTestEnv, Transport c) => TProxy c 'TServer -> (THandleSMP c 'TClient -> IO a) -> IO a + runClient _ = testSMPClient +``` + +Update the existing restore tests to read stats through `readSavedStats`: + +In `testRestoreMessages`, replace each of the three pairs + +```haskell + logSize testServerStatsBackupFile `shouldReturn` 101 + Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile +``` + +(and the same for `stats2`, `stats3`) with + +```haskell + stats1 <- readSavedStats msType +``` + +(`stats2`, `stats3` respectively), and replace `removeFile testServerStatsBackupFile` with `removeFileIfExists testServerStatsBackupFile`. + +In `testRestoreExpireMessages`, replace + +```haskell + Right ServerStatsData {_msgExpired} <- strDecode <$> B.readFile testServerStatsBackupFile +``` + +with + +```haskell + ServerStatsData {_msgExpired} <- readSavedStats msType +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/SMP server via TLS/Save server stats/"` +Expected: FAIL — memory: "should save stats while running" fails (`logSize` finds no file while running); both stores: "should count stored queues and messages" fails (`simplex_smp_messages_total 0` or `simplex_smp_queues_total1 0`); Postgres: `readSavedStats` gets "no saved stats". + +- [ ] **Step 3: Server imports (`Server.hs`)** + +After `import Simplex.Messaging.Server.Stats`: + +```haskell +import Simplex.Messaging.Server.Stats +import Simplex.Messaging.Server.StatsStore +``` + +Remove `renameFile` from `import UnliftIO.Directory (doesFileExist, renameFile)` if no other use remains (check with `rg -n "renameFile" src/Simplex/Messaging/Server.hs` after Step 7). + +In the existing `#if defined(dbServerPostgres)` import block: + +```haskell +#if defined(dbServerPostgres) +import Simplex.Messaging.Server.MsgStore.Postgres (getDbMessageStats) +import qualified Simplex.Messaging.Server.QueueStore.Postgres as PQ +#endif +``` + +- [ ] **Step 4: Start the save thread** + +In the `raceAny_` list: + +```haskell + <> serverStatsThread_ cfg + <> saveStatsThread_ cfg + <> prometheusMetricsThread_ cfg +``` + +After `serverStatsThread_` definition (before `logServerStats`): + +```haskell + saveStatsThread_ :: ServerConfig s -> [M s ()] + saveStatsThread_ ServerConfig {serverStatsBackupFile = Just _, statsSaveInterval} = [saveStatsThread statsSaveInterval] + saveStatsThread_ _ = [] + + saveStatsThread :: Int64 -> M s () + saveStatsThread interval = do + labelMyThread "saveStatsThread" + lock <- asks $ savingLock . server + forever $ do + liftIO $ threadDelay' $ 1000000 * interval + withLock' lock "saveStats" saveServerStats +``` + +- [ ] **Step 5: Save around the daily rollover (`logServerStats`)** + +Before `let interval = 1000000 * logInterval`: + +```haskell + lock <- asks $ savingLock . server + u <- askUnliftIO + let interval = 1000000 * logInterval +``` + +Replace `withFile statsFilePath AppendMode $ \h -> liftIO $ do` with + +```haskell + withFile statsFilePath AppendMode $ \h -> withLock' lock "logStats" $ liftIO $ do + unliftIO u saveServerStats + hSetBuffering h LineBuffering +``` + +(`hSetBuffering h LineBuffering` is the existing first line.) After the last statement of that block, the multi-line `T.hPutStrLn h $ T.intercalate "," (...)`, add at the same indentation: + +```haskell + unliftIO u saveServerStats +``` + +- [ ] **Step 6: Save stats before closing the store (`saveServer`)** + +```haskell + saveServer :: Bool -> M s () + saveServer drainMsgs = do + -- the database stats store is closed together with the message store + saveServerStats + ms <- asks msgStore_ + liftIO $ saveServerMessages drainMsgs ms >> closeMsgStore (fromMsgStore ms) + saveServerNtfs +``` + +- [ ] **Step 7: Replace `saveServerStats` and `restoreServerStats`** + +Replace both definitions with: + +```haskell +serverStatsStore :: M s (Maybe StatsStore) +serverStatsStore = do + ms <- asks msgStore_ + fmap (statsStore ms) <$> asks (serverStatsBackupFile . config) + where + statsStore :: MsgStore s -> FilePath -> StatsStore + statsStore = \case + StoreMemory _ -> StatsFile +#if defined(dbServerPostgres) + StoreDatabase ms -> StatsDB (PQ.dbStore $ queueStore ms) +#endif + +saveServerStats :: M s () +saveServerStats = do + ss <- asks serverStats + serverStatsStore >>= mapM_ (liftIO . (`saveStats` ss)) + +restoreServerStats :: forall s. MsgStoreClass s => Maybe MessageStats -> MessageStats -> M s () +restoreServerStats msgStats_ ntfStats = do + s <- asks serverStats + serverStatsStore >>= mapM_ (liftIO . tryAny . loadStats >=> restore s) + st <- asks msgStore + EntityCounts {queueCount} <- liftIO $ getEntityCounts @(StoreQueue s) $ queueStore st + liftIO $ do + writeIORef (qCount s) queueCount + writeIORef (msgCount s) $ maybe 0 storedMsgsCount msgStats_ + writeIORef (ntfCount s) $ storedMsgsCount ntfStats + atomicModifyIORef'_ (msgExpired s) (+ maybe 0 expiredMsgsCount msgStats_) + atomicModifyIORef'_ (msgNtfExpired s) (+ expiredMsgsCount ntfStats) + where + restore s = \case + Right (Just d) -> liftIO (setServerStats s d) >> logNote "server stats restored" + Right Nothing -> pure () + Left e -> do + logError $ "error restoring server stats: " <> tshow e + liftIO exitFailure +``` + +`restoreServerStats` keeps its call site (`restoreServerStats msgStats_ ntfStats`, ~:182). Check that `renameFile` and `B.readFile`/`strDecode` imports are still used elsewhere in `Server.hs`; remove only unused ones. + +- [ ] **Step 8: Build without warnings** + +Run: `cabal build -fserver_postgres --ghc-options=-O0 all` and `cabal build --ghc-options=-O0 lib:simplexmq` +Expected: no warnings in changed files. + +- [ ] **Step 9: Run server tests** + +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/SMP server via TLS/"` +Expected: 0 failures for memory and postgres-only stores, including "Save server stats", "Restore messages", "Restore messages (old / v2)". +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Core tests/Server stats tests/"` +Expected: 13 examples, 0 failures. +Run: `cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match="/Notifications server/"` +Expected: 0 failures (uses `withSmpServerStoreLogOn` with stats persistence). + +- [ ] **Step 10: Commit** + +```bash +git add src/Simplex/Messaging/Server.hs tests/ServerTests.hs +git commit -m "smp server: save stats periodically" +``` + +```json:metadata +{"files": ["src/Simplex/Messaging/Server.hs", "tests/ServerTests.hs"], "verifyCommand": "cabal test -fserver_postgres --ghc-options=-O0 --test-show-details=streaming --test-option=--match=\"/SMP server via TLS/\"", "acceptanceCriteria": ["restore via StatsStore per mode", "load errors exit", "store counts recomputed every start", "periodic save under savingLock", "rollover saves before and after daily line", "shutdown saves before closing store", "existing and new server tests pass"], "modelTier": "frontier"} +```