DuckDB CPU attribution: dedicated executor + KurrentDB integration

docs/server/diagnostics/metrics.md

@@ -393,6 +393,14 @@ kurrentdb_proc_contention_count_total 297 1688147136862
 kurrentdb_gc_pause_duration_max_seconds{range="16-20 seconds"} 0.0485873 1688147136862
 ```
 
+### DuckDB
+
+CPU time consumed by the dedicated DuckDB executor that owns every thread DuckDB runs on (used by secondary indexing and the query engine). Because the executor owns those threads, this is the DuckDB share of the process CPU reported by `kurrentdb_proc_cpu`: compare `rate(kurrentdb_duckdb_cpu_seconds_total)` against `kurrentdb_proc_cpu` to see the fraction of KurrentDB's CPU spent inside DuckDB.
+
+| Time Series | Type | Description |
+|:------------|:-----|:------------|
+| `kurrentdb_duckdb_cpu_seconds_total{role=<worker\|dispatcher>}` | [Counter](#common-types) | Cumulative CPU seconds consumed by DuckDB's executor threads, split by `role`: `worker` (DuckDB's parallel task-scheduler threads) and `dispatcher` (threads that run queries, reads, and index commits). Requires per-thread CPU support (Linux/Windows); reads 0 on macOS. Excludes DuckDB work the schema registry runs on its own threads (low volume; its parallel portions still run on worker threads). |
+
 ### Projections
 
 Projection metrics track the statistics for projections.

src/Directory.Packages.props

@@ -68,8 +68,8 @@
     <PackageVersion Include="Kurrent.Surge.Core" Version="1.1.1-alpha.1.71" />
     <PackageVersion Include="Kurrent.Surge.DataProtection" Version="1.1.1-alpha.1.71" />
     <PackageVersion Include="Kurrent.Surge.DuckDB" Version="1.1.1-alpha.1.71" />
-    <PackageVersion Include="Kurrent.Quack" Version="0.0.0-alpha.181" />
-    <PackageVersion Include="Kurrent.Quack.Arrow" Version="0.0.0-alpha.181" />
+    <PackageVersion Include="Kurrent.Quack" Version="0.0.0-local.3" />
+    <PackageVersion Include="Kurrent.Quack.Arrow" Version="0.0.0-local.3" />
     <PackageVersion Include="KurrentDB.Client" Version="1.0.0" />
     <PackageVersion Include="librdkafka.redist" Version="2.5.0" />
     <PackageVersion Include="LruCacheNet" Version="1.2.0" />

src/KurrentDB.Core.Testing/Helpers/MiniNode.cs

@@ -22,7 +22,6 @@
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Certificates;
 using KurrentDB.Core.Configuration.Sources;
-using KurrentDB.Core.DuckDB;
 using KurrentDB.Core.Messages;
 using KurrentDB.Core.Services.Monitoring;
 using KurrentDB.Core.Services.Storage;
@@ -270,11 +269,6 @@ public MiniNode(string pathname,
 		builder.Services.AddSerilog();
 		Node.Startup.ConfigureServices(builder.Services);
 		_webHost = builder.Build();
-		_webHost.Use(async (ctx, next) => {
-			var factory = ctx.RequestServices.GetRequiredService<DuckDBConnectionPoolLifetime>();
-			ctx.Features.Set<ConnectionScopedDuckDBConnectionPool>(new(factory));
-			await next();
-		});
 		Node.Startup.Configure(_webHost);
 		_started = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
 		_adminUserCreated = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

src/KurrentDB.Core/Bus/Extensions/PublisherReadExtensions.cs

@@ -255,7 +255,6 @@ IAsyncEnumerator<ReadResponse> GetEnumerator() {
 					user: SystemAccounts.System,
 					requiresLeader: false,
 					expiryStrategy: DefaultExpiryStrategy.Instance,
-					pool: null,
 					cancellationToken: cancellationToken
 				)
 				: new Enumerator.ReadIndexBackwards(
@@ -266,7 +265,6 @@ IAsyncEnumerator<ReadResponse> GetEnumerator() {
 					user: SystemAccounts.System,
 					requiresLeader: false,
 					expiryStrategy: DefaultExpiryStrategy.Instance,
-					pool: null,
 					cancellationToken: cancellationToken
 				);
 		}

src/KurrentDB.Core/Bus/Extensions/PublisherSubscribeExtensions.cs

@@ -80,7 +80,6 @@ public static async IAsyncEnumerable<ReadResponse> SubscribeToIndex(this IPublis
 			checkpoint: position,
 			user: SystemAccounts.System,
 			requiresLeader: false,
-			pool: null,
 			cancellationToken: cancellationToken
 		);
 

src/KurrentDB.Core/ClusterVNodeStartup.cs

@@ -137,9 +137,6 @@ public void Configure(WebApplication app) {
 		app.UseAuthorization();
 		app.UseAntiforgery();
 
-		// provides a lazy DuckDB connection pool unique to the connection
-		app.UseDuckDb();
-
 		// allow all subsystems to register their legacy controllers before calling MapLegacyHttp
 		foreach (var component in _plugableComponents)
 			component.ConfigureApplication(app, _configuration);
@@ -314,7 +311,10 @@ public void ConfigureServices(IServiceCollection services) {
 		services.AddGrpcReflection();
 #endif
 
-		services.AddDuckDb();
+		services.AddDuckDb(
+			_metricsConfiguration.ServiceName,
+			workerCount: _configuration.GetValue("KurrentDB:DuckDB:WorkerThreads", Math.Clamp(Environment.ProcessorCount / 2, 2, 16)),
+			dispatcherCount: _configuration.GetValue("KurrentDB:DuckDB:DispatcherThreads", Math.Clamp(Environment.ProcessorCount / 2, 2, 8)));
 
 		// Ask the node itself to add DI registrations
 		_configureNodeServices(services);

src/KurrentDB.Core/DuckDB/DuckDBCpuMetrics.cs

@@ -0,0 +1,39 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+using Kurrent.Quack;
+
+namespace KurrentDB.Core.DuckDB;
+
+// Reports the cumulative CPU time consumed by the DuckDB executor's worker and dispatcher threads as an
+// observable counter, tagged by role. This is the whole point of routing DuckDB work through the executor:
+// all DuckDB CPU is now attributable to threads the executor owns, so a single SampleCpu() reading sums them.
+public class DuckDBCpuMetrics {
+	public const string MeterName = "KurrentDB.DuckDB";
+
+	// Held so the meter — and therefore its ObservableCounter — stays rooted for the lifetime of this object
+	// (owned by DuckDBExecutorLifetime, i.e. process lifetime). Without this reference the meter could be
+	// garbage-collected and the instrument would silently stop reporting.
+	private readonly Meter _meter;
+
+	public DuckDBCpuMetrics(Meter meter, string serviceName, Func<IReadOnlyList<DuckDBExecutor.CpuSample>> sampleCpu) {
+		_meter = meter;
+		meter.CreateObservableCounter($"{serviceName}.duckdb.cpu.seconds", Observe,
+			description: "CPU time consumed by DuckDB's executor worker and dispatcher threads, in seconds " +
+				"(excludes DuckDB work SchemaRegistry runs on its own threads via a shared connection pool; " +
+				"that work's parallel portions still run on the executor's worker threads).");
+
+		IEnumerable<Measurement<double>> Observe() {
+			double workers = 0, dispatchers = 0;
+			foreach (var s in sampleCpu())
+				if (s.Role == "worker") workers += s.CpuSeconds; else dispatchers += s.CpuSeconds;
+			yield return new(workers, new KeyValuePair<string, object?>("role", "worker"));
+			yield return new(dispatchers, new KeyValuePair<string, object?>("role", "dispatcher"));
+		}
+	}
+}

src/KurrentDB.Core/DuckDB/DuckDBExecutorLifetime.cs

@@ -0,0 +1,111 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using DotNext;
+using Kurrent.Quack;
+using KurrentDB.Core.TransactionLog.Chunks;
+using KurrentDB.DuckDB;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace KurrentDB.Core.DuckDB;
+
+// Owns the DuckDB executor: database open (with thread + memory settings applied at open),
+// one-time schema setups, the shutdown checkpoint, and executor disposal.
+public sealed class DuckDBExecutorLifetime : Disposable, IHostedService {
+	private readonly ILogger<DuckDBExecutorLifetime> _log;
+	[CanBeNull] private string _tempPath;
+
+	public DuckDBExecutor Executor { get; }
+
+	// The CPU metric is created here (eagerly, at node startup) rather than via an activated DI singleton so the
+	// instrument exists even without a metrics consumer, and so it samples the very executor this lifetime owns.
+	public DuckDBCpuMetrics CpuMetrics { get; }
+
+	public DuckDBExecutorLifetime(
+		TFChunkDbConfig config,
+		IEnumerable<IDuckDBSetup> setups,
+		int workerCount,
+		int dispatcherCount,
+		string serviceName,
+		[CanBeNull] ILogger<DuckDBExecutorLifetime> log) {
+		_log = log ?? NullLogger<DuckDBExecutorLifetime>.Instance;
+
+		var path = config.InMemDb ? GetTempPath() : $"{config.Path}/kurrent.ddb";
+		var memoryMib = (int)(GC.GetGCMemoryInfo().TotalAvailableMemoryBytes / 1024 / 1024 * 0.25); // 25% of RAM, as before
+		var connectionString = $"Data Source={path};memory_limit={memoryMib}MB";
+
+		Executor = new DuckDBExecutor(connectionString, workerCount, dispatcherCount);
+		try {
+			CpuMetrics = new DuckDBCpuMetrics(new Meter(DuckDBCpuMetrics.MeterName, "1.0.0"), serviceName, Executor.SampleCpu);
+			_log.LogInformation("DuckDB executor started at {path}: {workers} workers, {dispatchers} dispatchers, memory_limit {memory}MB",
+				path, workerCount, dispatcherCount, memoryMib);
+
+			// One-time setups (IndexingDbSchema, SchemaDbSchema) — effects are database-wide, so running
+			// them once on any executor-owned connection preserves today's semantics exactly.
+			Executor.Execute(connection => {
+				foreach (var setup in setups)
+					setup.Execute(connection);
+				return 0;
+			}, CancellationToken.None).AsTask().GetAwaiter().GetResult();
+		} catch {
+			// A setup failure (e.g. a schema-migration error) throws before this object exists for anyone to
+			// Dispose(), so the executor's worker/dispatcher threads and open DB handle would leak. Dispose it.
+			Executor.DisposeAsync().AsTask().GetAwaiter().GetResult();
+			throw;
+		}
+
+		return;
+
+		string GetTempPath() {
+			_tempPath = Path.GetTempFileName();
+			File.Delete(_tempPath); // DuckDB refuses a pre-existing empty file; recreate at the same path
+			return _tempPath;
+		}
+	}
+
+	public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+
+	// Bound on the shutdown checkpoint: it runs on a dispatcher, and if streaming queries have saturated the
+	// dispatcher pool the checkpoint could otherwise block shutdown indefinitely. A missed checkpoint is not data
+	// loss — DuckDB replays its WAL on next open — so time out and continue rather than hang teardown. (The broader
+	// dispatcher-starvation question is tracked for the load/soak: size DispatcherThreads above the expected
+	// concurrent-streaming-query count, or split streaming onto its own dispatcher pool.)
+	private static readonly TimeSpan CheckpointTimeout = TimeSpan.FromSeconds(30);
+
+	public async Task StopAsync(CancellationToken cancellationToken) {
+		_log.LogDebug("Checkpointing DuckDB");
+		var checkpoint = Executor.Execute(connection => {
+			connection.Checkpoint();
+			return 0;
+		}, cancellationToken).AsTask();
+		try {
+			await checkpoint.WaitAsync(CheckpointTimeout, cancellationToken);
+		} catch (TimeoutException) {
+			_log.LogWarning("DuckDB shutdown checkpoint did not complete within {Timeout}; skipping it (the WAL is " +
+				"replayed on next open). This can happen if streaming queries have saturated the DuckDB dispatcher pool.",
+				CheckpointTimeout);
+		}
+	}
+
+	protected override void Dispose(bool disposing) {
+		if (disposing) {
+			Executor.DisposeAsync().AsTask().GetAwaiter().GetResult();
+			if (_tempPath != null) {
+				try {
+					File.Delete(_tempPath);
+				} catch (IOException) {
+					// let the OS clean it up
+				}
+			}
+		}
+		base.Dispose(disposing);
+	}
+}