feat(CosmosStore.Linq): Indexed Queries

CHANGELOG.md

@@ -8,6 +8,14 @@ The `Unreleased` section name is replaced by the expected version of next releas
 
 ## [Unreleased]
 
+### Added
+
+- `Equinox.CosmosStore.Linq`: Add LINQ querying support for Indexed `u`nfolds [#450](https://github.com/jet/equinox/pull/450)
+
+### Changed
+### Removed
+### Fixed
+
 <a name="4.1.0"></a>
 ## [4.1.0] - 2025-06-12
 

samples/Store/Integration/LogIntegration.fs

@@ -38,6 +38,7 @@ module EquinoxCosmosInterop =
             | Log.Metric.Query (Direction.Backward,c,m) -> "CosmosQueryB", m, Some c, m.ru
             | Log.Metric.QueryResponse (Direction.Forward,m) -> "CosmosResponseF", m, None, m.ru
             | Log.Metric.QueryResponse (Direction.Backward,m) -> "CosmosResponseB", m, None, m.ru
+            | Log.Metric.Index m -> "CosmosLinq", m, None, m.ru
             | Log.Metric.SyncSuccess m -> "CosmosSync200", m, None, m.ru
             | Log.Metric.SyncConflict m -> "CosmosSync409", m, None, m.ru
             | Log.Metric.SyncResync m -> "CosmosSyncResync", m, None, m.ru

src/Equinox.CosmosStore.Prometheus/CosmosStorePrometheus.fs

@@ -112,6 +112,7 @@ type LogSink(customTags: seq<string * string>) =
                 | Op       (Operation.Tip304,   m) -> observeTip  ("R", "query",    "tip",           "ok", "304") m
                 | Op       (Operation.Query,    m) -> observe     ("R", "query",    "query",         "ok")        m
                 | QueryRes (_direction,         m) -> observeRes  ("R", "query",    "queryPage")                  m
+                | Op       (Operation.Index,    m) -> observe     ("R", "linq",     "query",         "ok")        m
                 | Op       (Operation.Write,    m) -> observe     ("W", "transact", "sync",          "ok")        m
                 | Op       (Operation.Conflict, m) -> observe     ("W", "transact", "conflict",      "conflict")  m
                 | Op       (Operation.Resync,   m) -> observe     ("W", "transact", "resync",        "conflict")  m

src/Equinox.CosmosStore.Prometheus/Equinox.CosmosStore.Prometheus.fsproj

@@ -9,8 +9,10 @@
   </ItemGroup>
 
   <ItemGroup>
-    <ProjectReference Condition=" '$(Configuration)' == 'Debug' " Include="..\Equinox.CosmosStore\Equinox.CosmosStore.fsproj" />
-    <PackageReference Condition=" '$(Configuration)' == 'Release' " Include="Equinox.CosmosStore" Version="[4.0.0, 5.0.0)" />
+    <!--    <ProjectReference Condition=" '$(Configuration)' == 'Debug' " Include="..\Equinox.CosmosStore\Equinox.CosmosStore.fsproj" />-->
+    <!--    <PackageReference Condition=" '$(Configuration)' == 'Release' " Include="Equinox.CosmosStore" Version="[4.2.0, 5.0.0)" />-->
+    <!-- For now, use a `>=` as a hack. The minute we publish CosmosStore 4.2.0, change back to the normal scheme as above -->
+    <ProjectReference Include="..\Equinox.CosmosStore\Equinox.CosmosStore.fsproj" />
   </ItemGroup>
 
   <ItemGroup>

src/Equinox.CosmosStore/CosmosStore.fs

@@ -237,6 +237,8 @@ module Log =
         | Delete of Measurement
         /// Trimmed the Tip
         | Trim of Measurement
+        /// Queried via the Index; count=-1 -> aggregate operation
+        | Index of Measurement
     let [<return: Struct>] (|MetricEvent|_|) (logEvent: Serilog.Events.LogEvent): Metric voption =
         let mutable p = Unchecked.defaultof<_>
         logEvent.Properties.TryGetValue(PropertyTag, &p) |> ignore
@@ -265,7 +267,7 @@ module Log =
     let internal eventLen (x: #IEventData<EncodedBody>) = EncodedBody.jsonUtf8Bytes x.Data + EncodedBody.jsonUtf8Bytes x.Meta + 80
     let internal batchLen = Seq.sumBy eventLen
     [<RequireQualifiedAccess>]
-    type Operation = Tip | Tip404 | Tip304 | Query | Write | Resync | Conflict | Prune | Delete | Trim
+    type Operation = Tip | Tip404 | Tip304 | Query | Index | Write | Resync | Conflict | Prune | Delete | Trim
     let (|Op|QueryRes|PruneRes|) = function
         | Metric.Tip s                        -> Op (Operation.Tip, s)
         | Metric.TipNotFound s                -> Op (Operation.Tip404, s)
@@ -274,6 +276,8 @@ module Log =
         | Metric.Query (_, _, s)              -> Op (Operation.Query, s)
         | Metric.QueryResponse (direction, s) -> QueryRes (direction, s)
 
+        | Metric.Index s                      -> Op (Operation.Index, s)
+
         | Metric.SyncSuccess s                -> Op (Operation.Write, s)
         | Metric.SyncResync s                 -> Op (Operation.Resync, s)
         | Metric.SyncConflict s               -> Op (Operation.Conflict, s)
@@ -308,6 +312,7 @@ module Log =
                 let epoch = System.Diagnostics.Stopwatch.StartNew()
                 member val internal Tip = Counters() with get, set
                 member val internal Read = Counters() with get, set
+                member val internal Index = Counters() with get, set
                 member val internal Write = Counters() with get, set
                 member val internal Resync = Counters() with get, set
                 member val internal Conflict = Counters() with get, set
@@ -335,6 +340,7 @@ module Log =
                                                                                 epoch.Tip.Ingest m
                             | Op (Operation.Query,            BucketMsRu m)  -> epoch.Read.Ingest m
                             | QueryRes (_direction,          _)              -> ()
+                            | Op (Operation.Index,            BucketMsRu m)  -> epoch.Index.Ingest m
                             | Op (Operation.Write,            BucketMsRu m)  -> epoch.Write.Ingest m
                             | Op (Operation.Conflict,         BucketMsRu m)  -> epoch.Conflict.Ingest m
                             | Op (Operation.Resync,           BucketMsRu m)  -> epoch.Resync.Ingest m
@@ -351,13 +357,14 @@ module Log =
             let stats =
               [|nameof res.Tip,         res.Tip
                 nameof res.Read,        res.Read
+                nameof res.Index,       res.Index
                 nameof res.Write,       res.Write
                 nameof res.Resync,      res.Resync
                 nameof res.Conflict,    res.Conflict
                 nameof res.Prune,       res.Prune
                 nameof res.Delete,      res.Delete
                 nameof res.Trim,        res.Trim |]
-            let isRead = function nameof res.Tip | nameof res.Read | nameof res.Prune -> true | _ -> false
+            let isRead = function nameof res.Tip | nameof res.Read | nameof res.Index | nameof res.Prune | nameof res.Resync -> true | _ -> false
             let buckets = stats |> Seq.collect (fun (_n, stat) -> stat.Buckets) |> Seq.distinct |> Seq.sort |> Seq.toArray
             if Array.isEmpty buckets then () else