ingestion: OTel decode/write histograms, per-partition Kafka concurrency, redactJSON early-exit (#412)

* ingestion: OTel histograms for decode + write phase

Initialises a global MeterProvider in Start.startApp (nested inside the
existing TracerProvider bracket so shutdown order is meter→tracer, both
flushing pending exports via shutdown*Provider). Picks up the same
OTEL_EXPORTER_OTLP_ENDPOINT / headers as traces by default, with the
usual per-signal overrides.

Adds Pkg.Metrics with two top-level NOINLINE instruments
(monoscope.ingest.decode.duration, monoscope.ingest.write.duration, ms)
plus a `timed` helper that brackets the action so latency is recorded
on exception too — useful for the dual-write where a timeout is still
latency we want in the distribution.

ProcessMessage.processMessages now records the decode phase manually
and wraps insertAndHandOff in `timed`.

* Auto-format code with fourmolu

* ingestion: per-partition Kafka group concurrency

Group records by (topic, partition) instead of just topic and process
groups via pooledForConcurrentlyN with a new kafkaGroupConcurrency knob
(default 4). Each tpsFor still yields exactly one TP, so a failed group
only stalls its own partition's commits.

Bound stays well under the hasql pool (shared with web): the failure
mode to avoid is AcquisitionTimeout → Hasql.isTransientException → no
commit → redelivery storm. Single committer (the poll thread) — no
concurrent commitPartitionsOffsets.

* ingestion: symmetric Metrics.timed for decode + write, redactJSON elem idiom

Wraps the decode block in Metrics.timed (replacing manual getMonotonicTime
+ recordMs) so both phases use the same bracket-on-exception path. Drops
the GHC.Clock import.

redactJSON: switches `any T.null ps` to `"" \`elem\` ps` — equivalent but
reads as "have we consumed all path segments?" which matches intent.

Removes TODO.md — belongs in PR/issue tracker, not the repo.

* hpack: pick up Pkg.Metrics in test-dev other-modules

After merging master (which added the test-dev stanza in #411), the
auto-generated other-modules list was missing Pkg.Metrics — the stanza
was hpack-generated before this branch's new module existed.

* queue: use HashMap consistently for byTP + tpsFor, drop Data.Map import

The rest of Pkg/Queue.hs uses HM throughout; (Text, Int32) and Int32
keys are both Hashable, so HashMap is a drop-in. tpsFor's group is
always single-partition, so the map has size 1 — ordering doesn't
matter.

* queue: revert HashMap swap, KP.PartitionId has no Hashable instance

The previous 79bd1bc assumed crPartition was Int32 (Hashable), but
it's KP.PartitionId (Ord-only). Data.Map.Strict was load-bearing, not
an inconsistency. Comment added to prevent the next reviewer from
trying the same swap.

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: tonyalaribe

monoscope.cabal

@@ -200,6 +200,7 @@ library
       Pkg.ErrorMetrics
       Pkg.ExtractionWorker
       Pkg.Mail
+      Pkg.Metrics
       Pkg.Parser
       Pkg.Parser.Expr
       Pkg.Parser.Stats
@@ -801,6 +802,7 @@ test-suite test-dev
       Pkg.ErrorMetrics
       Pkg.ExtractionWorker
       Pkg.Mail
+      Pkg.Metrics
       Pkg.Parser
       Pkg.Parser.Expr
       Pkg.Parser.Stats

src/Pkg/Metrics.hs

@@ -0,0 +1,73 @@
+-- | Shared OpenTelemetry metric instruments.
+--
+-- Instruments are spec-safe to cache once and reuse globally, so we hold
+-- them as top-level 'unsafePerformIO' thunks rather than threading a
+-- 'Meter' through 'AuthContext'. The global 'MeterProvider' is initialised
+-- in 'Start.startApp' before any of these are forced; before init they
+-- silently no-op via the SDK's 'noopMeterProvider'.
+module Pkg.Metrics (
+  ingestDecodeHist,
+  ingestWriteHist,
+  recordMs,
+  timed,
+) where
+
+import Effectful (Eff, IOE, (:>))
+import Effectful.Exception (bracket)
+import GHC.Clock (getMonotonicTime)
+import OpenTelemetry.Attributes (Attribute, Attributes, emptyAttributes, unsafeAttributesFromListIgnoringLimits)
+import OpenTelemetry.Metric.Core (
+  Histogram (..),
+  Meter (..),
+  defaultAdvisoryParameters,
+  getGlobalMeterProvider,
+  getMeter,
+ )
+import Relude
+import System.IO.Unsafe (unsafePerformIO)
+
+
+monoscopeMeter :: Meter
+monoscopeMeter = unsafePerformIO $ getGlobalMeterProvider >>= flip getMeter "monoscope"
+{-# NOINLINE monoscopeMeter #-}
+
+
+-- | Wall-time spent decoding/redacting/building span vectors in
+-- 'ProcessMessage.processMessages', per batch.
+ingestDecodeHist :: Histogram
+ingestDecodeHist =
+  unsafePerformIO
+    $ meterCreateHistogram monoscopeMeter "monoscope.ingest.decode.duration" (Just "ms") Nothing defaultAdvisoryParameters
+{-# NOINLINE ingestDecodeHist #-}
+
+
+-- | Wall-time spent in the PG + TF dual write
+-- ('Telemetry.insertAndHandOff'), per batch.
+ingestWriteHist :: Histogram
+ingestWriteHist =
+  unsafePerformIO
+    $ meterCreateHistogram monoscopeMeter "monoscope.ingest.write.duration" (Just "ms") Nothing defaultAdvisoryParameters
+{-# NOINLINE ingestWriteHist #-}
+
+
+recordMs :: MonadIO m => Histogram -> Double -> [(Text, Attribute)] -> m ()
+recordMs h v attrs = liftIO $ histogramRecord h v (toAttrs attrs)
+
+
+-- | Time an action and record its wall-time (ms) on @hist@. Uses
+-- 'bracket' so the measurement fires on both success and exception —
+-- useful when you want failures represented in the distribution (e.g.
+-- DB write latency where a timeout is still latency). Pass @[]@ for
+-- no attributes.
+timed :: IOE :> es => Histogram -> [(Text, Attribute)] -> Eff es a -> Eff es a
+timed hist attrs act =
+  bracket
+    (liftIO getMonotonicTime)
+    (\t0 -> liftIO getMonotonicTime >>= \t1 -> recordMs hist ((t1 - t0) * 1000) attrs)
+    (const act)
+
+
+toAttrs :: [(Text, Attribute)] -> Attributes
+toAttrs [] = emptyAttributes
+toAttrs xs = unsafeAttributesFromListIgnoringLimits xs
+{-# INLINE toAttrs #-}

src/Pkg/Queue.hs

@@ -3,7 +3,6 @@ module Pkg.Queue (pubsubService, kafkaService, publishJSONToKafka, publishToDead
 import Control.Exception.Annotated (checkpoint)
 import Control.Lens ((^?), _Just)
 import Control.Lens qualified as L
-import Control.Monad.Extra (concatMapM)
 import Control.Monad.Trans.Resource (runResourceT)
 import Data.Aeson qualified as AE
 import Data.Annotation (toAnnotation)
@@ -35,6 +34,7 @@ import System.IO.Unsafe (unsafePerformIO)
 import System.Tracing (SpanStatus (..), addEvent, setStatus, withSpan)
 import System.Types (ATBackgroundCtx, runBackground)
 import UnliftIO (throwIO)
+import UnliftIO.Async (pooledForConcurrentlyN)
 import UnliftIO.Exception (bracket, try, tryAny)
 import UnliftIO.MVar (modifyMVar)
 
@@ -239,16 +239,12 @@ kafkaService appLogger appCtx tp kafkaTopics batchSize fn = checkpoint "kafkaSer
                 $ LogBase.logAttention "Kafka poll returned errors"
                 $ AE.object ["error_count" AE..= length leftRecords, "errors" AE..= map show leftRecords]
 
-              -- Group by topic: a single poll can interleave records from multiple
-              -- subscribed topics; each group needs its own ce-type and attributes.
-              -- Per-partition commit: only topic-groups whose batch succeeded
-              -- (or whose poison content was successfully DLQ'd) contribute to
-              -- the offset commit. Failed groups leave their partitions alone;
-              -- librdkafka redelivers them next poll, surfacing the outage as
-              -- consumer-group lag on those partitions only — healthy
-              -- partitions advance unaffected.
-              let byTopic = foldr (\r -> HM.insertWith (<>) r.crTopic.unTopicName (r :| [])) HM.empty rightRecords
-                  processGroup (topic, neRecords@(recc :| _)) = do
+              -- Group per (topic, partition) so groups can run concurrently
+              -- below while each tpsFor still yields exactly one TP — failed
+              -- groups stall only their own partition's commits.
+              -- Map (not HashMap): KP.PartitionId has Ord but no Hashable instance.
+              let byTP = foldr (\r -> Map.insertWith (<>) (r.crTopic.unTopicName, r.crPartition) (r :| [])) Map.empty rightRecords
+                  processGroup ((topic, _partition), neRecords@(recc :| _)) = do
                     let headers = consumerRecordHeadersToHashMap recc
                         ceType = ceTypeFor appCtx.config.kafkaDeadLetterTopic topic headers
                         attributes = HM.insert "ce-type" ceType headers
@@ -293,7 +289,11 @@ kafkaService appLogger appCtx tp kafkaTopics batchSize fn = checkpoint "kafkaSer
                               pure $ case dlqRes of
                                 Right _ -> successTps
                                 Left _ -> []
-              tps <- concatMapM processGroup (HM.toList byTopic)
+              -- Bound must stay well under the hasql pool (shared with web) —
+              -- AcquisitionTimeout (→ Hasql.isTransientException → no commit →
+              -- redelivery storm) is the failure mode to avoid. Single committer
+              -- (this thread) below — no concurrent commitPartitionsOffsets.
+              tps <- concat <$> pooledForConcurrentlyN appCtx.config.kafkaGroupConcurrency (Map.toList byTP) processGroup
 
               -- No poll-thread backoff: lag growth is the outage signal under
               -- per-partition commits. A reintroduced `threadDelay` here races

src/ProcessMessage.hs

@@ -57,6 +57,7 @@ import Models.Projects.Projects qualified as Projects
 import Models.Telemetry.Telemetry (Context (trace_state), OtelLogsAndSpans (..), generateSummary)
 import Models.Telemetry.Telemetry qualified as Telemetry
 import Pkg.DeriveUtils (AesonText (..), UUIDId (..), unAesonTextMaybe)
+import Pkg.Metrics qualified as Metrics
 import Pkg.SchemaLearning.Catalog qualified as Catalog
 import Pkg.SchemaLearning.Catalog qualified as Fields
 import Pkg.SchemaLearning.Hot qualified as SchemaHot
@@ -138,45 +139,40 @@ processMessages
 processMessages [] _ = pure []
 processMessages msgs attrs = do
   appCtx <- Eff.ask @AuthContext
-  rMsgs <-
-    catMaybes <$> forM msgs \(ackId, msg) -> case AE.eitherDecodeStrict (BS.filter (/= 0) msg) of
-      Left err -> Nothing <$ Log.logAttention "Error parsing json msgs" (object ["AckId" .= ackId, "Error" .= err, "OriginalMsg" .= decodeUtf8 @Text msg])
-      Right m -> pure $ Just (ackId, BS.length msg, m)
-
-  if null rMsgs
-    then pure []
-    else do
-      projectCaches <-
-        HM.fromList <$> liftIO do
-          -- Use parallel evaluation for cache fetching
-          let projectIds = (\(_, _, m) -> UUIDId m.projectId) <$> rMsgs
-          cachePairs <- forM projectIds $ \pid ->
-            Cache.fetchWithCache appCtx.projectCache pid $ \pid' -> do
-              mpjCache <- Projects.projectCacheByIdIO appCtx.hasqlJobsPool pid'
-              pure $! fromMaybe Projects.defaultProjectCache mpjCache
-          pure $! zip projectIds cachePairs
-
-      spans <- forM rMsgs \(rmAckId, rawSize, msg) -> do
-        let pid = UUIDId msg.projectId
-            !msgSize = fromIntegral rawSize
-        case HM.lookup pid projectCaches of
-          Just cache ->
-            let !totalDailyEvents = fromIntegral cache.dailyEventCount + fromIntegral cache.dailyMetricCount
-                !isFreeTier = cache.paymentPlan == "Free"
-                !hasExceededLimit = isFreeTier && totalDailyEvents >= freeTierDailyMaxEvents
-             in if hasExceededLimit
-                  then pure Nothing
-                  else do
-                    spanId <- UUID.genUUID
-                    trId <- UUID.toText <$> UUID.genUUID
-                    let !span' = convertRequestMessageToSpan msg msgSize (spanId, trId)
-                    pure (Just span')
-          Nothing -> pure Nothing
-
-      let !spanVec = V.fromList (catMaybes spans)
-      Telemetry.insertAndHandOff appCtx.env.enableTimefusionWrites appCtx.extractionWorker projectCaches spanVec
-
-      pure $ map fst3 rMsgs
+  (rMsgs, mWrite) <- Metrics.timed Metrics.ingestDecodeHist [] do
+    rMsgs <-
+      catMaybes <$> forM msgs \(ackId, msg) -> case AE.eitherDecodeStrict (BS.filter (/= 0) msg) of
+        Left err -> Nothing <$ Log.logAttention "Error parsing json msgs" (object ["AckId" .= ackId, "Error" .= err, "OriginalMsg" .= decodeUtf8 @Text msg])
+        Right m -> pure $ Just (ackId, BS.length msg, m)
+    if null rMsgs
+      then pure (rMsgs, Nothing)
+      else do
+        projectCaches <-
+          liftIO $ HM.fromList <$> forM (ordNub $ (\(_, _, m) -> UUIDId m.projectId) <$> rMsgs) \pid -> do
+            cache <-
+              Cache.fetchWithCache appCtx.projectCache pid
+                $ fmap (fromMaybe Projects.defaultProjectCache)
+                . Projects.projectCacheByIdIO appCtx.hasqlJobsPool
+            pure (pid, cache)
+
+        spans <- forM rMsgs \(rmAckId, rawSize, msg) -> runMaybeT do
+          let pid = UUIDId msg.projectId
+              !msgSize = fromIntegral rawSize
+          cache <- hoistMaybe $ HM.lookup pid projectCaches
+          let !totalDailyEvents = fromIntegral cache.dailyEventCount + fromIntegral cache.dailyMetricCount
+              !isFreeTier = cache.paymentPlan == "Free"
+          guard $ not (isFreeTier && totalDailyEvents >= freeTierDailyMaxEvents)
+          spanId <- lift UUID.genUUID
+          trId <- lift $ UUID.toText <$> UUID.genUUID
+          pure $! convertRequestMessageToSpan msg msgSize (spanId, trId)
+
+        pure (rMsgs, Just (projectCaches, V.fromList (catMaybes spans)))
+
+  forM_ mWrite \(projectCaches, spanVec) ->
+    Metrics.timed Metrics.ingestWriteHist []
+      $ Telemetry.insertAndHandOff appCtx.env.enableTimefusionWrites appCtx.extractionWorker projectCaches spanVec
+
+  pure $ map fst3 rMsgs
 
 
 -- | Process a single span to extract entities for hash-stamping.
@@ -631,6 +627,10 @@ instance {-# OVERLAPPING #-} AE.FromJSON (Either Text [Text]) where
 
 -- | Walk the JSON once, redact any fields which are in the list of json paths to be redacted.
 --
+-- Empty paths short-circuits to structural sharing: subtrees with no live path
+-- are returned by pointer instead of rebuilt. Key matching requires a @.@
+-- boundary (tightening over a prior prefix-match implementation).
+--
 -- >>> redactJSON (V.fromList ["menu.id."]) [aesonQQ| {"menu":{"id":"file", "name":"John"}} |]
 -- Object (fromList [("menu",Object (fromList [("id",String "[REDACTED]"),("name",String "John")]))])
 --
@@ -642,19 +642,33 @@ instance {-# OVERLAPPING #-} AE.FromJSON (Either Text [Text]) where
 --
 -- >>> redactJSON (V.fromList ["menu.[].id", "menu.[].names.[]"]) [aesonQQ| {"menu":[{"id":"i1", "names":["John","okon"]}, {"id":"i2"}]} |]
 -- Object (fromList [("menu",Array [Object (fromList [("id",String "[REDACTED]"),("names",Array [String "[REDACTED]",String "[REDACTED]"])]),Object (fromList [("id",String "[REDACTED]")])])])
+--
+-- Prefix without a path-boundary does NOT match:
+--
+-- >>> redactJSON (V.fromList ["password"]) [aesonQQ| {"pass":{"word":"secret"}} |]
+-- Object (fromList [("pass",Object (fromList [("word",String "secret")]))])
 redactJSON :: V.Vector Text -> AE.Value -> AE.Value
-redactJSON paths' = redactJSON' (map stripPrefixDot $ V.toList paths')
+redactJSON ps0 = go [fromMaybe p (T.stripPrefix "." p) | p <- V.toList ps0]
   where
-    redactJSON' !paths value = case value of
-      AET.String v -> if any T.null paths then AET.String "[REDACTED]" else AET.String v
-      AET.Number v -> if any T.null paths then AET.String "[REDACTED]" else AET.Number v
-      AET.Null -> AET.Null
-      AET.Bool v -> AET.Bool v
-      AET.Object objMap -> AET.Object $ AEKM.mapWithKey (\k v -> redactJSON' (matchKey (AEK.toText k) paths) v) objMap
-      AET.Array jsonList -> AET.Array $ V.map (redactJSON' (mapMaybe (\path -> T.stripPrefix "[]." path <|> T.stripPrefix "[]" path) paths)) jsonList
-
-    matchKey !k = mapMaybe (\path -> T.stripPrefix (k <> ".") path <|> T.stripPrefix k path)
-    stripPrefixDot !p = fromMaybe p (T.stripPrefix "." p)
+    go [] v = v
+    go ps v = case v of
+      AET.Object om -> AET.Object $ AEKM.mapWithKey (\k -> go (mapMaybe (matchKey (AEK.toText k)) ps)) om
+      AET.Array xs
+        | null cps -> v
+        | otherwise -> AET.Array $ V.map (go cps) xs
+        where
+          cps = mapMaybe (\p -> T.stripPrefix "[]." p <|> T.stripPrefix "[]" p) ps
+      AET.String{} | "" `elem` ps -> AET.String "[REDACTED]"
+      AET.Number{} | "" `elem` ps -> AET.String "[REDACTED]"
+      _ -> v
+
+    -- Match @k@ against a path on a @.@ boundary, returning the remainder.
+    -- Avoids allocating @k <> "."@ per (key, path) pair.
+    matchKey !k path =
+      T.stripPrefix k path >>= \rest -> case T.uncons rest of
+        Nothing -> Just ""
+        Just ('.', rest') -> Just rest'
+        _ -> Nothing
 
 
 -- valueToFields takes an aeson object and converts it into a vector of paths to

src/Start.hs

@@ -5,6 +5,7 @@ where
 
 import Configuration.Dotenv qualified as Dotenv
 import Control.Exception.Safe qualified as Safe
+import OpenTelemetry.Metric (initializeGlobalMeterProvider, shutdownMeterProvider)
 import OpenTelemetry.Trace (Tracer, TracerOptions (..), getGlobalTracerProvider, initializeGlobalTracerProvider, makeTracer, shutdownTracerProvider)
 import Relude
 import System.Server qualified as Server
@@ -13,11 +14,23 @@ import System.Server qualified as Server
 startApp :: IO ()
 startApp = do
   _ <- Safe.try (Dotenv.loadFile Dotenv.defaultConfig) :: IO (Either SomeException ())
-  withTracer \tracer -> getGlobalTracerProvider >>= Server.runMonoscope
+  withTracer \_tracer ->
+    withMeter $ getGlobalTracerProvider >>= Server.runMonoscope
   where
     withTracer :: ((TracerOptions -> Tracer) -> IO c) -> IO c
     withTracer f =
       Safe.bracket
         initializeGlobalTracerProvider
         (\tp -> void $ shutdownTracerProvider tp Nothing)
         (\tracerProvider -> f $ makeTracer tracerProvider "monoscope-server")
+
+    -- Metric instruments live in 'Pkg.Metrics' and pull from the global
+    -- meter provider lazily; init here so the global is set before the
+    -- first 'histogramRecord'. Reads OTEL_METRICS_EXPORTER / interval env
+    -- vars (defaults: otlp, 60s).
+    withMeter :: IO c -> IO c
+    withMeter action =
+      Safe.bracket
+        initializeGlobalMeterProvider
+        (\mp -> void $ shutdownMeterProvider mp Nothing)
+        (\_ -> action)

src/System/Config.hs

@@ -56,6 +56,7 @@ data EnvConfig = EnvConfig
   , kafkaUsername :: Text
   , kafkaPassword :: Text
   , enableKafkaService :: Bool
+  , kafkaGroupConcurrency :: Int
   , smtpHost :: Text
   , smtpPort :: Int
   , smtpTls :: Bool
@@ -213,6 +214,7 @@ instance DefConfig EnvConfig where
       , postmarkFromEmail = "hello@monoscope.tech"
       , openaiModel = "gpt-5.4-mini"
       , openaiSmallModel = "gpt-5.4-nano"
+      , kafkaGroupConcurrency = 4
       , extractionWorkerShards = 4
       , extractionQueueCapacity = 64
       , drainFlushBatchSize = 1000