scoped service

Author: KSemenenko

AGENTS.md

@@ -12,6 +12,9 @@
 - When initialization logic is required (initialization tasks), add a simple hosted service to perform it and register that hosted service inside the relevant extensions.
 - Do not expose user-facing knobs for AGE/Postgres connection pool sizing (e.g., `MaxConnections` parameters); rely on EF Core-style connection-string keywords and reasonable defaults inside the connector.
 - Graph store registrations (Postgres, Neo4j, Cosmos) must automatically register a default `IGraphStore`; remove `MakeDefault` toggles/options and rely on the first registration when an unkeyed graph store is requested.
+- Keep bulk graph store operations on `IGraphStore`; do not split them into a separate `IBulkGraphStore` interface.
+- Avoid separate scoped graph store abstractions (e.g., `IScopedGraphStore`); keep scope management on the primary client/graph store or its factory rather than exposing an extra DI service.
+- Always update `README.md` (and related docs) to reflect any behavior or API changes you make so documentation stays current with the code.
 
 # Conversations
 any resulting updates to agents.md should go under the section "## Rules to follow"

Directory.Build.props

@@ -4,7 +4,8 @@
     <LangVersion>14</LangVersion>
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
-      <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
+    <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
+    <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
 
     <!--NuGet-->
@@ -25,8 +26,8 @@
         <RepositoryUrl>https://github.com/managedcode/graphrag</RepositoryUrl>
         <PackageProjectUrl>https://github.com/managedcode/graphrag</PackageProjectUrl>
         <Product>Managed Code GraphRag</Product>
-        <Version>10.0.2</Version>
-        <PackageVersion>10.0.2</PackageVersion>
+        <Version>10.0.3</Version>
+        <PackageVersion>10.0.3</PackageVersion>
 
     </PropertyGroup>
     <PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">

README.md

@@ -236,7 +236,7 @@ GraphRAG ships with a first-class Apache AGE adapter (`ManagedCode.GraphRag.Post
      }
    }
    ```
-4. **Register through DI.** `services.AddPostgresGraphStore("postgres", configure: ...)` wires up `IAgeConnectionManager`, `IAgeClientFactory`, `IGraphStore`, `IScopedGraphStore`, and `IBulkGraphStore`. Pool sizing follows the standard Npgsql settings (configure `Max Pool Size`, `Timeout`, etc. inside the connection string). The first registration becomes the default unkeyed `IGraphStore`; additional stores remain keyed-only.
+4. **Register through DI.** `services.AddPostgresGraphStore("postgres", configure: ...)` wires up `IAgeConnectionManager`, `IAgeClientFactory`, `PostgresGraphStore`, `IGraphStore`, and `PostgresExplainService`. Pool sizing follows the standard Npgsql settings (configure `Max Pool Size`, `Timeout`, etc. inside the connection string). The first registration becomes the default unkeyed `IGraphStore`; additional stores remain keyed-only.
 
    ```csharp
    var services = new ServiceCollection()
@@ -253,23 +253,27 @@ GraphRAG ships with a first-class Apache AGE adapter (`ManagedCode.GraphRag.Post
    var graphStore = provider.GetRequiredService<IGraphStore>();
 
    // Scoped operations reuse a single AGE/Postgres connection for the lifetime of the scope
-   var scopedStore = provider.GetRequiredService<IScopedGraphStore>();
-   await using (await scopedStore.CreateScopeAsync())
+   var ageClientFactory = provider.GetRequiredKeyedService<IAgeClientFactory>("postgres");
+   await using (await ageClientFactory.CreateScopeAsync())
    {
        await graphStore.UpsertNodeAsync("node-1", "Example", new Dictionary<string, object?> { ["name"] = "Scoped" });
        await graphStore.UpsertNodeAsync("node-2", "Example", new Dictionary<string, object?> { ["name"] = "Connection" });
    }
 
    // Bulk helpers batch large workloads while keeping the scoped connection alive
-   var bulkStore = provider.GetRequiredService<IBulkGraphStore>();
-   await bulkStore.UpsertNodesAsync(new[]
+   await graphStore.UpsertNodesAsync(new[]
    {
        new GraphNodeUpsert("bulk-1", "Example", new Dictionary<string, object?> { ["name"] = "Bulk" }),
        new GraphNodeUpsert("bulk-2", "Example", new Dictionary<string, object?> { ["name"] = "Write" })
    });
+
+   await graphStore.UpsertRelationshipsAsync(new[]
+   {
+       new GraphRelationshipUpsert("bulk-1", "bulk-2", "RELATES_TO", new Dictionary<string, object?>())
+   });
    ```
 
-   The `AgeConnectionManager` automatically retries transient `53300: too many clients` errors (up to three exponential backoff attempts) so scopes can wait for a free slot before failing. When a scope is disposed, the underlying `IAgeClientScope` returns its connection to the pool, keeping concurrency predictable even under heavy fan-out.
+   The `AgeConnectionManager` automatically retries transient `53300: too many clients` errors (up to three exponential backoff attempts) so scopes can wait for a free slot before failing. When a scope is disposed, the underlying `IAgeClientScope` created by `IAgeClientFactory` returns its connection to the pool, keeping concurrency predictable even under heavy fan-out.
 
 ### Neo4j Setup
 

src/ManagedCode.GraphRag.CosmosDb/CosmosGraphStore.cs

@@ -32,6 +32,16 @@ public async Task UpsertNodeAsync(string id, string label, IReadOnlyDictionary<s
         await container.UpsertItemAsync(document, new PartitionKey(document.Id), cancellationToken: cancellationToken);
     }
 
+    public async Task UpsertNodesAsync(IReadOnlyCollection<GraphNodeUpsert> nodes, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(nodes);
+
+        foreach (var node in nodes)
+        {
+            await UpsertNodeAsync(node.Id, node.Label, node.Properties, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
     public async Task UpsertRelationshipAsync(string sourceId, string targetId, string type, IReadOnlyDictionary<string, object?> properties, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
@@ -43,6 +53,31 @@ public async Task UpsertRelationshipAsync(string sourceId, string targetId, stri
         await container.UpsertItemAsync(document, new PartitionKey(document.SourceId), cancellationToken: cancellationToken);
     }
 
+    public async Task UpsertRelationshipsAsync(IReadOnlyCollection<GraphRelationshipUpsert> relationships, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(relationships);
+
+        foreach (var relationship in relationships)
+        {
+            await UpsertRelationshipAsync(
+                relationship.SourceId,
+                relationship.TargetId,
+                relationship.Type,
+                relationship.Properties,
+                cancellationToken).ConfigureAwait(false);
+
+            if (relationship.Bidirectional)
+            {
+                await UpsertRelationshipAsync(
+                    relationship.TargetId,
+                    relationship.SourceId,
+                    relationship.Type,
+                    relationship.Properties,
+                    cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
     public IAsyncEnumerable<GraphRelationship> GetOutgoingRelationshipsAsync(string sourceId, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);

src/ManagedCode.GraphRag.Neo4j/Neo4jGraphStore.cs

@@ -33,6 +33,16 @@ public async Task UpsertNodeAsync(string id, string label, IReadOnlyDictionary<s
         await session.ExecuteWriteAsync(tx => tx.RunAsync(cypher, new { id, props = properties }));
     }
 
+    public async Task UpsertNodesAsync(IReadOnlyCollection<GraphNodeUpsert> nodes, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(nodes);
+
+        foreach (var node in nodes)
+        {
+            await UpsertNodeAsync(node.Id, node.Label, node.Properties, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
     public async Task UpsertRelationshipAsync(string sourceId, string targetId, string type, IReadOnlyDictionary<string, object?> properties, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);
@@ -49,6 +59,31 @@ public async Task UpsertRelationshipAsync(string sourceId, string targetId, stri
         await session.ExecuteWriteAsync(tx => tx.RunAsync(cypher, new { sourceId, targetId, props = properties }));
     }
 
+    public async Task UpsertRelationshipsAsync(IReadOnlyCollection<GraphRelationshipUpsert> relationships, CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(relationships);
+
+        foreach (var relationship in relationships)
+        {
+            await UpsertRelationshipAsync(
+                relationship.SourceId,
+                relationship.TargetId,
+                relationship.Type,
+                relationship.Properties,
+                cancellationToken).ConfigureAwait(false);
+
+            if (relationship.Bidirectional)
+            {
+                await UpsertRelationshipAsync(
+                    relationship.TargetId,
+                    relationship.SourceId,
+                    relationship.Type,
+                    relationship.Properties,
+                    cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
     public IAsyncEnumerable<GraphRelationship> GetOutgoingRelationshipsAsync(string sourceId, CancellationToken cancellationToken = default)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(sourceId);

src/ManagedCode.GraphRag.Postgres/ApacheAge/AgeClientFactory.cs

@@ -7,6 +7,7 @@ namespace GraphRag.Storage.Postgres.ApacheAge;
 public interface IAgeClientFactory
 {
     IAgeClient CreateClient();
+    IAgeClient CreateUnscopedClient();
     ValueTask<IAgeClientScope> CreateScopeAsync(CancellationToken cancellationToken = default);
 }
 
@@ -26,6 +27,8 @@ public IAgeClient CreateClient()
         return CreatePhysicalClient();
     }
 
+    public IAgeClient CreateUnscopedClient() => CreatePhysicalClient();
+
     public ValueTask<IAgeClientScope> CreateScopeAsync(CancellationToken cancellationToken = default)
     {
         cancellationToken.ThrowIfCancellationRequested();

src/ManagedCode.GraphRag.Postgres/ApacheAge/AgeConnectionManager.cs

@@ -41,6 +41,7 @@ public AgeConnectionManager(string connectionString, ILogger<AgeConnectionManage
             throw new ArgumentOutOfRangeException(nameof(connectionString), "Maximum Pool Size must be greater than zero.");
         }
         connectionBuilder.MinPoolSize = Math.Min(10, connectionBuilder.MaxPoolSize);
+        connectionBuilder.Timeout = 0;
 
         ConnectionString = connectionBuilder.ConnectionString;
         _dataSource = NpgsqlDataSource.Create(connectionBuilder.ConnectionString);
@@ -182,12 +183,23 @@ private async Task<NpgsqlConnection> OpenDataSourceConnectionAsync(CancellationT
                 LogMessages.ConnectionRetrying(_logger, ConnectionString, attempt, delay, ex.MessageText);
                 await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
             }
+            catch (NpgsqlException ex) when (ShouldRetryOnPoolExhaustion(ex, attempt))
+            {
+                var delay = GetRetryDelay(attempt);
+                LogMessages.ConnectionRetrying(_logger, ConnectionString, attempt, delay, ex.Message);
+                await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
+            }
         }
     }
 
     private static bool ShouldRetry(PostgresException ex, int attempt) =>
         ex.SqlState == PostgresErrorCodes.TooManyConnections && attempt < ConnectionLimitMaxAttempts;
 
+    private static bool ShouldRetryOnPoolExhaustion(NpgsqlException ex, int attempt) =>
+        attempt < ConnectionLimitMaxAttempts &&
+        ex.InnerException is TimeoutException &&
+        ex.Message.Contains("The connection pool has been exhausted", StringComparison.OrdinalIgnoreCase);
+
     private static TimeSpan GetRetryDelay(int attempt)
     {
         var delayMillis = Math.Min(